ThinkPHP 8.0 基于 Redis Stream 实现消息队列消费【MQ】-ThinkPHP-PHP中文网

thinkphp 8.0 原生队列不支持 redis stream，需手动封装；stream 相比 list 具备消费者组、xack 确认、xpending 监控、唯一消息 id 等核心优势，但要求严格实现 ack 机制与进程管理。

thinkphp 8.0 基于 redis stream 实现消息队列消费【mq】

ThinkPHP 8.0 原生队列驱动（think-queue）默认不支持 Redis Stream，它仅封装了 List（LPUSH/RPOP）和 Sorted Set（延迟队列）结构。若需使用 Redis Stream 实现更可靠、带消费者组、消息确认（ACK）、待处理查询（XPENDING）能力的 MQ，必须手动封装，不能依赖框架内置队列命令。

为什么选 Stream 而非 List？

Stream 是 Redis 5.0+ 引入的原生日志型数据结构，相比 List 队列有本质优势：

天然支持消费者组（Consumer Group）：多 worker 协作消费同一队列，自动负载均衡，避免重复拉取
消息确认机制（XACK）：只有成功处理并显式 ACK 的消息才会被标记为已消费，失败可重试或人工干预
待处理监控（XPENDING）：能查出卡住超时、未 ACK 的消息 ID 和所属消费者，便于故障定位
消息持久化与唯一 ID：每条消息自带时间戳+序列号（如 1712345678901-0001），可追溯、可按 ID 精确重放

关键配置与初始化逻辑

需在项目中自行封装一个 Stream 操作类，核心步骤包括：

连接 Redis 并指定 DB（如 select 2 专用于消息队列）
定义队列名（stream key）、消费者组名（group name）、消费者名称（consumer name）
首次运行时检查 stream 是否存在，不存在则创建；再检查消费者组是否存在，不存在则用 XGROUP CREATE <stream><group> 0</group></stream> 初始化（从头开始消费）
推荐设置 mkstream 参数，确保 stream 自动创建：$redis->xadd($streamKey, '*', $data, ['MKSTREAM' => true])

生产者：安全入队

不要直接用 rawCommand 拼接，应封装健壮方法：

btpanel phpsite 宝塔面板PHP网站

宝塔面板 PHP 网站管理：站点创建、删除、启停、PHP 版本切换、域名管理、SSL证书管理、伪静态管理、数据库管理

下载

消息体建议 JSON 编码，并包含必要上下文字段（如 trace_id、created_at、retry_count）
入队前可加简单校验（如非空、必填字段存在），避免脏数据污染 stream
调用 $redis->xadd($streamKey, '*', $msgArray) 返回消息 ID，可用于日志追踪或幂等写入
若需延迟投递，不能靠 Stream 原生命令实现，需配合定时任务扫描 + XADD 或改用 Sorted Set 做延迟层

消费者：闭环处理流程

TP8 中不能用 php think queue:work 启动 Stream 消费，需自建常驻脚本（如 app/command/StreamConsume.php），执行逻辑必须严格遵循：

使用 XREADGROUP GROUP <group><consumer> COUNT 10 BLOCK 2000 STREAMS <stream> ></stream></consumer></group> 拉取消息（> 表示只读新消息）
收到消息后，立即业务处理 —— 注意：处理中禁止抛未捕获异常，否则进程退出；应 try/catch 并记录错误
处理成功后，**必须调用 XACK**，否则该消息会持续出现在 XPENDING 列表中
处理失败且需重试时，可暂不 ACK，等待下次拉取；或主动 XCLAIM 抢回并修改 retry_count 后重新投递
建议每轮消费后加 usleep(10000) 防止单次循环过快打满 CPU