消费者组解决什么问题
Streams 的消费者组把一条消息分发给组内的多个消费者，同时记录谁在处理、处理了多久。消费者读取消息后必须用 XACK 确认，未确认的消息会留在待处理列表（PEL）里；消费者宕机后，其他消费者可以按空闲时间把消息认领回来重试。这套机制让「消费失败不丢消息」变成框架能力而不是业务自己造。
基本信息
| 适用平台 | Redis 5.0+ 的 Streams；XAUTOCLAIM 需 6.2+ |
|---|---|
| 建组命令 | XGROUP CREATE stream group id [MKSTREAM] |
| 读取命令 | XREADGROUP GROUP g consumer COUNT n BLOCK ms STREAMS key > |
| 确认命令 | XACK stream group id |
| 故障转移 | XAUTOCLAIM（6.2+）或 XCLAIM |
典型使用场景
- 订单异步处理：多个消费者并行消费，失败可重投
- 消息投递审计：确认哪些消息已被处理、哪些仍在挂起
- 消费者扩容缩容：新实例加入后自动分到未处理消息
- 延迟任务：结合空闲时间认领实现超时重试
- 需要保留历史消息并支持回溯读取的场景
完整消费流程
完整消费流程
XGROUP CREATE orders g1 0 MKSTREAM # 建组并从 0 开始（含历史消息）
XREADGROUP GROUP g1 consumer1 COUNT 10 BLOCK 2000 STREAMS orders >
# 处理完成后确认，未确认的消息会留在 PEL
XACK orders g1 1737100000000-0
# 查看待处理概览与明细
XPENDING orders g1
XPENDING orders g1 - + 10
# 认领空闲超过 60 秒的消息给新消费者
XAUTOCLAIM orders g1 consumer2 60000 0-0 COUNT 10
# 观察组与消费者状态
XINFO GROUPS orders
XINFO CONSUMERS orders g1关键命令与作用
| 命令 | 作用 | 要点 |
|---|---|---|
| XGROUP CREATE | 创建消费者组 | 指定起始 ID 决定是否消费历史 |
| XREADGROUP > | 读取从未投递的新消息 | 指定 ID 则读自己 PEL 中的历史 |
| XACK | 确认消息已处理 | 不确认会被重复投递 |
| XPENDING | 查看待处理消息 | 数量和空闲时间都要看 |
| XAUTOCLAIM | 按空闲时间批量认领 | 6.2+ 提供，替代逐个 XCLAIM |
| XINFO CONSUMERS | 查看消费者状态 | 已废弃的消费者会挂着 pending |
消费逻辑必须幂等：消息可能被多次投递，业务处理要以消息 ID 或业务唯一键做去重，这样重试才安全。
PEL 只增不减通常有两个原因：消费者没有执行 XACK，或者消费者已下线但没人认领它的消息。把 XPENDING 的数量纳入监控，才能及时发现消息堆积。