本文目录导读：
在PHP中实现延迟队列有几种主流方案,我来详细介绍：
基于Redis的延迟队列
使用Redis Sorted Set实现
<?php
class DelayQueue {
private $redis;
private $queueName;
public function __construct($queueName = 'delay_queue') {
$this->redis = new Redis();
$this->redis->connect('127.0.0.1', 6379);
$this->queueName = $queueName;
}
/**
* 添加延迟任务
* @param string $task 任务数据
* @param int $delay 延迟秒数
*/
public function addTask($task, $delay) {
$score = time() + $delay;
return $this->redis->zAdd($this->queueName, $score, $task);
}
/**
* 获取到期的任务
*/
public function getReadyTasks() {
$now = time();
// 获取所有已到期的任务
$tasks = $this->redis->zRangeByScore($this->queueName, 0, $now);
if (!empty($tasks)) {
// 移除已获取的任务
$this->redis->zRemRangeByScore($this->queueName, 0, $now);
}
return $tasks;
}
/**
* 检查任务是否到期
*/
public function getNextTask() {
$tasks = $this->getReadyTasks();
return empty($tasks) ? null : $tasks[0];
}
}
// 使用示例
$queue = new DelayQueue();
// 添加延迟任务
$queue->addTask(json_encode(['order_id' => 123, 'action' => 'cancel']), 300); // 5分钟后执行
// 消费任务
$task = $queue->getNextTask();
if ($task) {
$data = json_decode($task, true);
// 处理任务
echo "处理订单: " . $data['order_id'];
}
使用Redis List + 定时轮询
<?php
class SimpleDelayQueue {
private $redis;
public function __construct() {
$this->redis = new Redis();
$this->redis->connect('127.0.0.1', 6379);
}
/**
* 添加延迟任务
*/
public function push($queueName, $data, $delay = 0) {
$delayTime = time() + $delay;
$item = json_encode([
'data' => $data,
'execute_time' => $delayTime
]);
return $this->redis->lPush("queue:{$queueName}:waiting", $item);
}
/**
* 获取可用任务
*/
public function pop($queueName) {
$waitingKey = "queue:{$queueName}:waiting";
$readyKey = "queue:{$queueName}:ready";
// 移出到期的任务
while ($item = $this->redis->rPop($waitingKey)) {
$task = json_decode($item, true);
if ($task['execute_time'] <= time()) {
// 任务已到期，存入可执行队列
$this->redis->lPush($readyKey, json_encode($task['data']));
} else {
// 任务未到期，放回队列并退出
$this->redis->lPush($waitingKey, $item);
break;
}
}
// 返回可执行的任务
return $this->redis->rPop($readyKey);
}
}
基于消息队列（RabbitMQ）的延迟队列
安装依赖
composer require php-amqplib/php-amqplib
实现代码
<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
class RabbitMQDelayQueue {
private $connection;
private $channel;
public function __construct() {
$this->connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$this->channel = $this->connection->channel();
}
/**
* 投递延迟消息
* @param string $exchange 交换机
* @param string $routingKey 路由键
* @param array $data 消息数据
* @param int $delay 延迟毫秒数
*/
public function publish($exchange, $routingKey, $data, $delay) {
// 设置延迟属性
$headers = new \PhpAmqpLib\Wire\AMQPTable([
'x-delay' => $delay
]);
$msg = new AMQPMessage(json_encode($data), [
'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
]);
// 设置延迟插件需要的参数
$msg->set('application_headers', $headers);
// 声明延迟交换机
$this->channel->exchange_declare(
'delayed_exchange',
'x-delayed-message',
false,
true,
false,
false,
false,
new \PhpAmqpLib\Wire\AMQPTable([
'x-delayed-type' => 'direct'
])
);
$this->channel->basic_publish($msg, 'delayed_exchange', $routingKey);
}
/**
* 消费消息
*/
public function consume($queue, $callback) {
$this->channel->queue_declare($queue, false, true, false, false);
$this->channel->queue_bind($queue, 'delayed_exchange', 'delayed_routing_key');
$this->channel->basic_consume($queue, '', false, true, false, false, $callback);
while ($this->channel->is_consuming()) {
$this->channel->wait();
}
}
public function __destruct() {
$this->channel->close();
$this->connection->close();
}
}
// RabbitMQ需要安装延迟插件
// rabbitmq-plugins enable rabbitmq_delayed_message_exchange
使用数据库实现简单延迟队列
<?php
class DatabaseDelayQueue {
private $db;
public function __construct(PDO $db) {
$this->db = $db;
}
/**
* 创建延迟任务表
*/
public function createTable() {
$sql = "
CREATE TABLE IF NOT EXISTS delay_tasks (
id INT AUTO_INCREMENT PRIMARY KEY,
task_data TEXT NOT NULL,
execute_at INT NOT NULL,
status ENUM('pending', 'processing', 'completed', 'failed') DEFAULT 'pending',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_execute_at (execute_at),
INDEX idx_status (status)
)
";
$this->db->exec($sql);
}
/**
* 添加延迟任务
*/
public function addTask($data, $delay) {
$stmt = $this->db->prepare(
"INSERT INTO delay_tasks (task_data, execute_at) VALUES (?, ?)"
);
return $stmt->execute([
json_encode($data),
time() + $delay
]);
}
/**
* 获取可执行任务并锁定
*/
public function getAvailableTasks($limit = 10) {
$this->db->beginTransaction();
try {
// 使用行级锁或乐观锁
$stmt = $this->db->prepare(
"SELECT * FROM delay_tasks
WHERE execute_at <= ? AND status = 'pending'
ORDER BY execute_at ASC
LIMIT ? FOR UPDATE"
);
$stmt->execute([time(), $limit]);
$tasks = $stmt->fetchAll(PDO::FETCH_ASSOC);
if (!empty($tasks)) {
$ids = array_column($tasks, 'id');
$placeholders = implode(',', array_fill(0, count($ids), '?'));
$updateStmt = $this->db->prepare(
"UPDATE delay_tasks SET status = 'processing' WHERE id IN ($placeholders)"
);
$updateStmt->execute($ids);
}
$this->db->commit();
return $tasks;
} catch (Exception $e) {
$this->db->rollBack();
throw $e;
}
}
/**
* 完成任务
*/
public function completeTask($id) {
$stmt = $this->db->prepare(
"UPDATE delay_tasks SET status = 'completed' WHERE id = ?"
);
return $stmt->execute([$id]);
}
}
使用第三方库（推荐）
Laravel 队列
// Laravel 内置延迟队列支持
dispatch(new ProcessPodcast($podcast))
->delay(now()->addMinutes(10));
// 或使用 Job
ProcessPodcast::dispatch($podcast)
->onQueue('processing')
->delay(now()->addMinutes(10));
Yii2 队列扩展
// 使用 yiisoft/yii2-queue
class DelayTask extends \yii\base\BaseObject implements \yii\queue\JobInterface
{
public $data;
public function execute($queue)
{
// 处理任务
}
}
// 延迟执行
Yii::$app->queue->delay(1800)->push(new DelayTask([
'data' => 'delayed task'
]));
最佳实践建议
-
生产环境推荐方案
- 小规模项目：Redis Sorted Set
- 中大型项目：RabbitMQ + 延迟插件
- 微服务架构：Kafka + 时间轮
-
注意事项
- 任务幂等性处理
- 失败重试机制
- 数据持久化
- 监控告警
-
性能优化
- 使用批量处理
- 合理设置轮询间隔
- 避免数据库锁表
选择合适的方案要根据项目规模、团队技术栈和运维能力来决定。
