66 lines
1.8 KiB
PHP
66 lines
1.8 KiB
PHP
<?php
|
||
namespace app\common\service;
|
||
|
||
use Predis\Client as RedisClient;
|
||
|
||
class QueueService
|
||
{
|
||
protected $redis;
|
||
|
||
public function __construct()
|
||
{
|
||
$this->redis = new RedisClient([
|
||
'host' => '127.0.0.1',
|
||
'port' => 6379,
|
||
// 'password' => '', // 如果有密码
|
||
]);
|
||
}
|
||
|
||
// 推送消息到队列
|
||
public function push($queueName, $message)
|
||
{
|
||
$this->redis->rpush($queueName, json_encode($message));
|
||
}
|
||
|
||
// 从队列中取出消息
|
||
public function pop($queueName)
|
||
{
|
||
$message = $this->redis->lpop($queueName);
|
||
return $message ? json_decode($message, true) : null;
|
||
}
|
||
|
||
/**
|
||
* 延迟推送（ZSET）
|
||
* @param string $queueName
|
||
* @param array|string $data
|
||
* @param int $delay 延迟秒数
|
||
*/
|
||
public function laterPush($queueName, $data, $delay)
|
||
{
|
||
$score = time() + $delay;
|
||
$payload = is_string($data) ? $data : json_encode($data, JSON_UNESCAPED_UNICODE);
|
||
return $this->redis->zadd($queueName . ':delayed', [$payload => $score]);
|
||
}
|
||
|
||
/**
|
||
* 检查延迟队列，将到期任务迁移到普通队列
|
||
* 并发安全：先 zrem，只有「移除成功（返回>0）」的进程才把任务放进就绪队列，
|
||
* 避免多个消费者同时迁移导致同一任务被投递多次。
|
||
* @param string $queueName
|
||
*/
|
||
public function migrateDelayedToReady($queueName)
|
||
{
|
||
$now = time();
|
||
$tasks = $this->redis->zrangebyscore($queueName . ':delayed', 0, $now);
|
||
if (empty($tasks)) {
|
||
return;
|
||
}
|
||
foreach ($tasks as $task) {
|
||
// 先抢占式移除，再入队
|
||
if ($this->redis->zrem($queueName . ':delayed', $task) > 0) {
|
||
$this->push($queueName, $task);
|
||
}
|
||
}
|
||
}
|
||
}
|