PHPAPI延迟队列怎么实现

wen PHP项目 2026-07-05 214

本文目录导读：

PHPAPI延迟队列怎么实现

基于Redis的延迟队列
使用Redis List + 定时轮询
基于消息队列（RabbitMQ）的延迟队列
使用数据库实现简单延迟队列
使用第三方库（推荐）
最佳实践建议

在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 + 时间轮
注意事项
- 任务幂等性处理
- 失败重试机制
- 数据持久化
- 监控告警
性能优化
- 使用批量处理
- 合理设置轮询间隔
- 避免数据库锁表

选择合适的方案要根据项目规模、团队技术栈和运维能力来决定。

本文地址： https://www.bmjup.com/post/16651.html