64 lines
1.7 KiB
PHP
64 lines
1.7 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]);
|
||
}
|
||
|
||
/**
|
||
* 检查延迟队列，将到期任务迁移到普通队列
|
||
* @param string $queueName
|
||
*/
|
||
public function migrateDelayedToReady($queueName)
|
||
{
|
||
$now = time();
|
||
// 获取到期任务
|
||
$tasks = $this->redis->zrangebyscore($queueName . ':delayed', 0, $now);
|
||
if (!empty($tasks)) {
|
||
foreach ($tasks as $task) {
|
||
// 放回普通队列
|
||
$this->push($queueName, $task);
|
||
// 从延迟队列中移除
|
||
$this->redis->zrem($queueName . ':delayed', $task);
|
||
}
|
||
}
|
||
}
|
||
}
|