IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

PHP 延迟消息取消方案

wen PHP项目 2

本文目录导读：

PHP 延迟消息取消方案

  1. Redis + 延迟队列（推荐方案）
  2. Redis + 流（Stream）方案
  3. RabbitMQ 延迟消息方案
  4. 数据库定时扫描方案
  5. 其他高级方案
  6. 选择建议

在 PHP 中实现延迟消息的取消，主要有以下几种方案，我会从简单到复杂逐一介绍：

Redis + 延迟队列（推荐方案）

原理

使用 Redis 的有序集合（ZSet），通过 score 存储执行时间戳，通过一个独立进程扫描到期任务。

<?php
class DelayedQueue {
    private $redis;
    private $queueKey = 'delayed:queue';
    private $cancelKey = 'delayed:cancel';
    public function __construct($redis) {
        $this->redis = $redis;
    }
    /**
     * 添加延迟任务
     */
    public function add(string $taskId, array $data, int $delaySeconds): bool {
        $executeTime = time() + $delaySeconds;
        $task = json_encode([
            'id' => $taskId,
            'data' => $data,
            'create_time' => time()
        ]);
        // 将任务添加到 ZSet，score 为执行时间戳
        return $this->redis->zAdd($this->queueKey, $executeTime, $task);
    }
    /**
     * 取消延迟任务
     */
    public function cancel(string $taskId): bool {
        // 将所有该 taskId 的任务标记为取消
        $cancelData = json_encode([
            'id' => $taskId,
            'cancel_time' => time()
        ]);
        return $this->redis->hSet($this->cancelKey, $taskId, $cancelData);
    }
    /**
     * 检查任务是否被取消（消费者处理时调用）
     */
    public function isCancelled(string $taskId): bool {
        return $this->redis->hExists($this->cancelKey, $taskId);
    }
    /**
     * 获取到期任务（消费者轮询）
     */
    public function pollReadyTasks(): array {
        $now = time();
        $tasks = $this->redis->zRangeByScore(
            $this->queueKey, 
            '-inf', 
            $now,
            ['limit' => [0, 100]]
        );
        $readyTasks = [];
        foreach ($tasks as $taskJson) {
            $task = json_decode($taskJson, true);
            // 检查是否被取消
            if (!$this->isCancelled($task['id'])) {
                $readyTasks[] = $task;
                // 从队列中移除
                $this->redis->zRem($this->queueKey, $taskJson);
            } else {
                // 被取消的任务直接移除
                $this->redis->zRem($this->queueKey, $taskJson);
                $this->redis->hDel($this->cancelKey, $task['id']);
            }
        }
        return $readyTasks;
    }
    /**
     * 清理已过期的取消标记
     */
    public function cleanupCancelMarkers(int $maxAge = 86400): int {
        $count = 0;
        $cancelTasks = $this->redis->hGetAll($this->cancelKey);
        foreach ($cancelTasks as $taskId => $cancelData) {
            $data = json_decode($cancelData, true);
            if (time() - $data['cancel_time'] > $maxAge) {
                $this->redis->hDel($this->cancelKey, $taskId);
                $count++;
            }
        }
        return $count;
    }
}

消费者处理

<?php
// consumer.php - 独立进程运行
while (true) {
    $queue = new DelayedQueue($redis);
    // 获取到期的任务
    $readyTasks = $queue->pollReadyTasks();
    foreach ($readyTasks as $task) {
        try {
            // 执行任务
            processTask($task);
        } catch (Exception $e) {
            // 记录错误日志
            error_log("Task {$task['id']} failed: " . $e->getMessage());
        }
    }
    // 定期清理取消标记
    if (time() % 3600 == 0) {
        $queue->cleanupCancelMarkers();
    }
    // 每100毫秒轮询一次
    usleep(100000);
}

Redis + 流（Stream）方案

Redis 5.0+ 的 Stream 数据结构，支持消息确认和待处理列表。

<?php
class DelayedStreamQueue {
    private $redis;
    private $streamKey = 'task:stream';
    private $groupName = 'task:group';
    public function __construct($redis) {
        $this->redis = $redis;
    }
    /**
     * 添加延迟任务
     */
    public function add(string $taskId, array $data, int $delaySeconds): string {
        $executeTime = time() + $delaySeconds;
        $messageId = $this->redis->xAdd($this->streamKey, '*', [
            'task_id' => $taskId,
            'data' => json_encode($data),
            'execute_time' => $executeTime,
            'status' => 'pending'
        ]);
        return $messageId;
    }
    /**
     * 取消任务 - 标记为已取消
     */
    public function cancel(string $taskId): bool {
        // 使用 Lua 脚本来原子性更新所有匹配的任务
        $luaScript = <<<LUA
        local stream = KEYS[1]
        local taskId = ARGV[1]
        local cancelTime = ARGV[2]
        -- 获取流中的信息
        local entries = redis.call('XRANGE', stream, '-', '+')
        local updated = 0
        for _, entry in ipairs(entries) do
            local msgId = entry[1]
            local fields = entry[2]
            local task_id = fields[2] -- 这里需要根据实际结构调整
            if task_id == taskId then
                -- 更新状态为已取消
                redis.call('XACK', stream, KEYS[2], msgId)
                redis.call('XDEL', stream, msgId)
                updated = updated + 1
            end
        end
        return updated
        LUA;
        return $this->redis->eval($luaScript, 2, $this->streamKey, $this->groupName, $taskId, time());
    }
}

RabbitMQ 延迟消息方案

使用 TTL + 死信队列

<?php
class RabbitMQDelayedQueue {
    private $connection;
    private $channel;
    public function __construct($connection) {
        $this->connection = $connection;
        $this->channel = $connection->channel();
        $this->setup();
    }
    private function setup() {
        // 声明主队列（延迟队列）
        $this->channel->queue_declare(
            'delay_queue',     // 队列名
            false,             // passive
            true,              // durable
            false,             // exclusive
            false,             // auto_delete
            false,             // no_wait
            [
                'x-dead-letter-exchange' => ['exchange', 'task.exchange'],
                'x-dead-letter-routing-key' => ['routing_key', 'task.process'],
                'x-message-ttl' => ['I', 5000] // 5秒延迟
            ]
        );
        // 声明处理队列
        $this->channel->queue_declare(
            'process_queue',
            false,
            true,
            false,
            false
        );
        // 声明交换机
        $this->channel->exchange_declare('task.exchange', 'direct', false, true, false);
        // 绑定
        $this->channel->queue_bind('process_queue', 'task.exchange', 'task.process');
    }
    /**
     * 发布延迟消息
     */
    public function addTask(string $taskId, array $data, int $delayMs) {
        $this->channel->queue_declare(
            'delay_queue_' . $taskId,  // 每个任务一个队列来动态设置 TTL
            false,
            true,
            false,
            false,
            false,
            [
                'x-dead-letter-exchange' => ['exchange', 'task.exchange'],
                'x-dead-letter-routing-key' => ['routing_key', 'task.process'],
                'x-message-ttl' => ['I', $delayMs]
            ]
        );
        $message = new AMQPMessage(json_encode([
            'task_id' => $taskId,
            'data' => $data
        ]));
        $this->channel->basic_publish(
            $message,
            '',
            'delay_queue_' . $taskId
        );
    }
    /**
     * 取消延迟任务
     */
    public function cancelTask(string $taskId) {
        // 直接删除对应的延迟队列
        $this->channel->queue_delete('delay_queue_' . $taskId);
    }
}

数据库定时扫描方案

适用于中小型项目,使用数据库存储任务状态。

<?php
class DatabaseDelayQueue {
    private $pdo;
    /**
     * 添加延迟任务
     */
    public function add(string $taskId, array $data, int $delaySeconds): bool {
        $stmt = $this->pdo->prepare(
            'INSERT INTO delayed_tasks (task_id, data, execute_at, status, created_at)
             VALUES (?, ?, DATE_ADD(NOW(), INTERVAL ? SECOND), ?, NOW())'
        );
        return $stmt->execute([
            $taskId,
            json_encode($data, JSON_UNESCAPED_UNICODE),
            $delaySeconds,
            'pending'
        ]);
    }
    /**
     * 取消任务
     */
    public function cancel(string $taskId): bool {
        $stmt = $this->pdo->prepare(
            'UPDATE delayed_tasks 
             SET status = ?, cancelled_at = NOW()
             WHERE task_id = ? AND status = ?
             AND execute_at > NOW()'
        );
        return $stmt->execute(['cancelled', $taskId, 'pending']);
    }
    /**
     * 获取并执行到期任务
     */
    public function processDueTasks(): array {
        $now = new DateTime();
        $stmt = $this->pdo->prepare(
            'SELECT task_id, data 
             FROM delayed_tasks
             WHERE execute_at <= ? AND status = ?
             FOR UPDATE SKIP LOCKED'
        );
        $stmt->execute([$now->format('Y-m-d H:i:s'), 'pending']);
        $tasks = $stmt->fetchAll(PDO::FETCH_ASSOC);
        // 更新任务状态为执行中
        foreach ($tasks as $task) {
            $this->pdo->prepare(
                'UPDATE delayed_tasks SET status = ? WHERE task_id = ?'
            )->execute(['processing', $task['task_id']]);
        }
        return $tasks;
    }
    /**
     * 完成任务
     */
    public function complete(string $taskId): bool {
        $stmt = $this->pdo->prepare(
            'UPDATE delayed_tasks 
             SET status = ?, completed_at = NOW()
             WHERE task_id = ?'
        );
        return $stmt->execute(['completed', $taskId]);
    }
}

其他高级方案

使用 Redis RStream 扩展

// 使用 RedStream 或 RedisStreams 库
$stream = new RedisStreams\Stream($redis, 'tasks');
// 添加带延迟的任务
$stream->add([
    'task_id' => $taskId,
    'data' => json_encode($data),
    'execute_at' => time() + $delaySeconds
], $maxLen);
// 取消任务
$stream->query(
    RedisStreams\Query::greaterThan($lastId)
);

使用 PECL 插件

// pecl 安装 redis 扩展后
$queue = new \RedisQueue\DelayedQueue($redis);
$queue->addJob($taskId, $data, $delaySeconds);
$queue->cancelJob($taskId);

选择建议

方案 优点 缺点 适用场景
Redis ZSet 实现简单、性能高、支持高效取消 需要额外消费者进程 多数场景，推荐
Redis Stream 持久化、支持消费组、可追溯 实现复杂度较高 需要消息确认机制
RabbitMQ 专业 MQ、可靠性高、支持 TTL 部署复杂、资源占用大 企业级应用
数据库轮询 简单直观、事务支持好

上一篇PHP 事务消息怎么实现

下一篇PHP 怎么使用消息队列解耦

相关文章

抱歉，评论功能暂时关闭!