PHP用Redis做队列

PHP + Redis 实战指南：从零搭建高性能消息队列系统（附完整代码）

目录导读（Table of Contents）

1. 为什么用 Redis 做队列？——业务痛点与选型对比
2. 基础篇：Redis 队列的三种核心数据结构（List / ZSet / Stream）
3. 实战篇：PHP 封装一个稳定可靠的延迟队列（消费确认 + 失败重试）
4. 避坑指南：Redis 队列常见的 5 个性能陷阱与解决方案
5. 高频问答：Redis 队列，面试官最爱问的 4 个问题

为什么用 Redis 做队列？——业务痛点与选型对比

在 PHP 高并发场景（如秒杀、订单超时、异步邮件）中，传统 MySQL 表轮询会造成数据库压力过大，而 RabbitMQ/Kafka 又存在部署过重、维护成本高的问题。Redis 凭借单线程原子操作、毫秒级读写、支持持久化等特性，成为中小型项目的首选队列中间件。

选型对比：

• List 队列：基于 LPUSH/BRPOP，适合简单任务分发，但无法保证消息不丢失。
• ZSet 延迟队列：通过 score 存储时间戳，完美解决订单超时关单问题。
• Stream 队列：Redis 5.0+ 引入，支持消费者组、ACK 确认，最接近专业 MQ 的体验。

基础篇：Redis 队列的三种核心数据结构

(1) List 队列（实时任务）

// 生产者
$redis->lpush('task:email', json_encode(['to'=>'a@b.com']));
// 消费者（阻塞式，防止空轮询）
$data = $redis->brpop('task:email', 5); // 5秒超时

注意：BRPOP 能防止 CPU 空转，但必须设置超时时间。

(2) ZSet 延迟队列（定时任务）

// 延迟 30 分钟执行任务
$redis->zadd('delay:order', time() + 1800, $orderId);
// 循环取到期任务
$tasks = $redis->zRangeByScore('delay:order', 0, time());
foreach ($tasks as $id) {
    $redis->zrem('delay:order', $id); // 移除并处理
}

(3) Stream 队列（可靠消费）

// 生产者
$redis->xAdd('stream:pay', '*', ['order_id'=>1001]);
// 消费者组（保证消息只被一个 Worker 处理）
$redis->xGroup('CREATE', 'stream:pay', 'group1', '0', true);
$msg = $redis->xReadGroup('group1', 'consumer1', ['stream:pay'=>'>'], 1, 5000);
// 处理完成后必须 XACK，否则消息会重新进入 PEL
$redis->xAck('stream:pay', 'group1', [$msg['id']]);

实战篇：PHP 封装一个稳定可靠的延迟队列（消费确认 + 失败重试）

目标： 解决消息丢失、重复消费、失败重试三大问题。

核心思路： 使用 ZSet 做延迟调度，配合 List 做即时处理，双重保障。

class RedisDelayQueue {
    private $redis;
    private $delayKey = 'delay:queue';
    private $readyKey = 'ready:queue';
    private $retryKey = 'retry:queue';
    public function add($task, $delaySeconds = 0) {
        // 延迟任务进 ZSet，即时任务直接进 List
        if ($delaySeconds > 0) {
            $this->redis->zAdd($this->delayKey, time() + $delaySeconds, $task);
        } else {
            $this->redis->rPush($this->readyKey, $task);
        }
    }
    public function process() {
        // 1. 将到期任务从 ZSet 迁移到 List
        $dueTasks = $this->redis->zRangeByScore($this->delayKey, 0, time());
        foreach ($dueTasks as $task) {
            $this->redis->zRem($this->delayKey, $task);
            $this->redis->rPush($this->readyKey, $task);
        }
        // 2. 从 List 阻塞消费
        $task = $this->redis->lPop($this->readyKey);
        if (!$task) return;
        try {
            $this->handleTask($task);
        } catch (\Exception $e) {
            // 3. 失败重试（最多 3 次）
            $retryCount = $this->redis->hIncrBy($this->retryKey, $task, 1);
            if ($retryCount < 3) {
                $this->redis->rPush($this->readyKey, $task); // 放回队列
            } else {
                $this->redis->hDel($this->retryKey, $task);
                $this->logger->error("任务失败: " . $task);
            }
        }
    }
}

关键点： 使用 lPop（非阻塞）配合 process() 的死循环调度，方便控制并发；用 Hash 记录重试次数。

避坑指南：Redis 队列常见的 5 个性能陷阱与解决方案

高频问答：Redis 队列，面试官最爱问的 4 个问题

Q1：Redis List 做队列，如何防止消息丢失？

A：使用 BRPOPLPUSH source destination timeout 命令，原子地将待处理消息从主队列弹出并备份到备份队列，处理成功后删除备份，若 Worker 崩溃，可扫描备份队列重新投递。

Q2：Redis 和 RabbitMQ 的队列本质区别是什么？

A：Redis 是内存数据库，优先级在于速度（微秒级），但未确认消息丢失风险高；RabbitMQ 是专业的消息中介，支持持久化、路由、事务，吞吐量低于 Redis，但可靠性更强。建议： 日志监控用 Redis，交易核心用 RabbitMQ。

Q3：延迟队列如何实现精确到秒的触发？

A：ZSet 的 score 存储触发时间戳，每秒钟轮询一次 zRangeByScore 获取该秒到期的任务，如果任务量巨大（每秒>1万），可改用 Redis 6.0 的 keyspace notification 结合 SET 过期事件，但需要注意过期事件可能延迟。

Q4：如何让 Redis 队列的消费者负载均衡？

A：多个 Worker 进程使用 BRPOP 同一 List 时，Redis 会自动公平分配消息（因为底层是互斥弹栈），对于 Stream，使用消费者组 XReadGroup，每个组内消费者轮流获取消息。

使用 Redis 做队列，本质是在速度与可靠性之间做取舍，对于中小型 PHP 项目，List + ZSet 组合已能覆盖 90% 的异步场景，若业务涉及金融交易，建议替换为 RabbitMQ 或 Kafka，最后记住：任何队列都要有监控机制（LLEN、内存占用、消费延迟）。

参考资源： 本项目完整代码可复用上述示例，部署时注意 Redis 配置 maxmemory-policy noeviction 防止队列键被淘汰。