91 lines
3.0 KiB
PHP
91 lines
3.0 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;
|
||
}
|
||
|
||
/**
|
||
* 阻塞式取出消息：队列空时挂在 Redis 上等，有消息立刻返回。
|
||
*
|
||
* 相比 pop() + sleep(1) 的轮询，取到消息的延迟从「最多 1 秒」降到接近 0，
|
||
* 对「下单 -> 调上游发放」这种链路首段能实打实省掉一截等待。
|
||
*
|
||
* @param string $queueName
|
||
* @param int $timeout 最长阻塞秒数（到点返回 null，便于外层做兜底巡检）
|
||
* @return array|string|null
|
||
*/
|
||
public function bpop($queueName, $timeout = 5)
|
||
{
|
||
// BLPOP 会把连接挂住 $timeout 秒，必须短于 Redis 连接的读超时
|
||
// （未显式配置时取 PHP default_socket_timeout，通常 60 秒），
|
||
// 否则会抛读超时异常而不是正常返回 null。这里夹到 30 秒以内。
|
||
$timeout = max(1, min(30, (int) $timeout));
|
||
// Predis 的 blpop 返回 [队列名, 消息]，超时返回 null
|
||
$res = $this->redis->blpop([$queueName], $timeout);
|
||
if (empty($res[1])) {
|
||
return null;
|
||
}
|
||
$decoded = json_decode($res[1], true);
|
||
return $decoded === null ? $res[1] : $decoded;
|
||
}
|
||
|
||
/**
|
||
* 延迟推送（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);
|
||
}
|
||
}
|
||
}
|
||
}
|