4. Redis Stream 详解：数据结构、消费者组与消息确认机制
Redis Stream 是 Redis 5.0 引入的原生消息队列数据结构，它解决了 Pub/Sub 无持久化、List 消息确认困难、ZSet 无阻塞消费等核心痛点。Stream 以日志追加模型存储消息，支持消费者组实现消息的负载均衡与可靠投递，是构建生产级消息队列的最佳选择。
4.1 Stream 核心数据结构
Stream 本质上是一个仅追加的日志，每条消息由唯一的 ID 和 键值对 组成。消息 ID 默认由服务器自动生成，格式为 <毫秒时间戳>-<序列号>，保证全局有序且唯一。
| 概念 | 说明 | 示例 |
|---|---|---|
| 消息 ID | 全局唯一、单调递增，支持按 ID 范围查询 | 1712345678000-0 |
| 消息体 | 由多个 field-value 对组成，类似 Hash | task_id:1001, status:new |
| Stream Key | Redis 键，存储整个消息流 | order:stream |
| 阻塞读取 | 支持 XREAD 阻塞等待新消息 |
XREAD BLOCK 5000 |
关键命令速览：
XADD key * field value [field value ...]— 追加消息XRANGE key start end— 按 ID 范围查询XREAD COUNT n BLOCK ms STREAMS key id— 阻塞读取XLEN key— 获取消息长度
4.2 消费者组（Consumer Group）
消费者组是 Stream 实现消息负载均衡与可靠消费的核心机制。一个消费者组包含多个消费者，每条消息只会被组内的一个消费者消费，且消费状态由 Redis 服务端维护。
消费者组 vs 普通 XREAD：普通 XREAD 是广播模式，所有消费者都会收到相同消息；消费者组实现的是“点对点”模式，每条消息只投递给一个消费者。
4.2.1 消费者组核心概念
| 概念 | 说明 | 命令/属性 |
|---|---|---|
| 组名 | 消费者组的唯一标识 | XGROUP CREATE |
| 消费者 | 组内的消费实例，可动态加入/退出 | 自动注册 |
| 待处理列表（PEL） | 已投递但未确认的消息列表 | XPENDING |
| last_delivered_id | 组内最后投递的消息 ID | 服务端维护 |
| 消费者偏移量 | 每个消费者已读取的位置 | 独立维护 |
4.2.2 消费者组工作流程
- 创建组：
XGROUP CREATE mystream mygroup $（$ 表示从最新消息开始消费） - 消费者读取：
XREADGROUP GROUP mygroup consumer1 COUNT 1 BLOCK 2000 STREAMS mystream > - 消息投递：服务端将消息分配给组内某个消费者，并记录到 PEL
- 消息确认：
XACK mystream mygroup message-id，从 PEL 中移除 - 失败重试：通过
XPENDING查看未确认消息，使用XCLAIM转移所有权
注意：消费者组使用
> 符号表示“只读取从未投递给任何消费者的新消息”。如果指定具体 ID，则用于读取历史消息或故障恢复。
4.3 消息确认机制（ACK）
消息确认是保证至少一次投递语义的关键。消费者处理完消息后，必须显式调用 XACK 通知 Redis 服务端，否则消息会一直留在 PEL 中，成为待处理消息。
4.3.1 未确认消息的处理
- 查看待处理消息：
XPENDING mystream mygroup返回 PEL 中的消息数量、最小/最大 ID 及每个消费者的待处理数 - 查看详情：
XPENDING mystream mygroup - + 10列出具体消息 ID、所属消费者、空闲时间、投递次数 - 消息认领：
XCLAIM mystream mygroup consumer2 60000 message-id将超时未确认的消息转移给其他消费者 - 自动认领：
XAUTOCLAIM mystream mygroup consumer2 60000 0-0自动扫描并认领超时消息
最佳实践：设置合理的
ACK 超时时间（如 60 秒），结合 XAUTOCLAIM 实现故障自动恢复。避免 PEL 无限增长导致内存泄漏。
4.3.2 消息投递语义对比
| 语义 | 实现方式 | 适用场景 |
|---|---|---|
| 最多一次 | XREAD 读取后不确认 | 日志采集、可丢失监控 |
| 至少一次 | XREADGROUP + XACK | 订单处理、任务调度 |
| 精确一次 | XREADGROUP + XACK + 业务幂等 | 支付回调、金融交易 |
4.4 任务去重方案：Stream + 幂等性
由于 Stream 保证至少一次投递，消费者可能收到重复消息。常见的去重方案包括：
- 业务 ID 去重：在消息体中携带全局唯一 ID（如
task_id），消费者处理前检查是否已处理 - Redis Set 去重：使用
SADD记录已处理 ID，设置过期时间防止内存膨胀 - Stream 消息 ID 去重：利用消息 ID 的唯一性，结合
XACK状态判断
推荐方案：使用
SETNX 或 SET EX NX 实现分布式锁去重，配合 Stream 的 PEL 机制，即使消费者崩溃重启也不会丢失消息。
4.5 生产级配置建议
| 配置项 | 推荐值 | 说明 |
|---|---|---|
| 消息最大长度 | MAXLEN ~ 100000 |
使用近似裁剪，防止内存无限增长 |
| 消费者组数量 | 1~3 个 | 每个业务场景一个组，避免组过多导致性能下降 |
| ACK 超时时间 | 60~120 秒 | 根据业务处理耗时设定，超时后自动认领 |
| PEL 大小监控 | 告警阈值 10000 | PEL 过大说明消费者处理不及时或存在死信 |
| 消息体大小 | < 1 KB | 避免大消息阻塞 Stream 写入 |
避坑指南：
- 不要使用
XGROUP CREATE的MKSTREAM参数自动创建空 Stream，可能导致消费者组指向不存在的 ID - 消费者崩溃后，未 ACK 的消息会一直留在 PEL，需定期使用
XAUTOCLAIM清理 - Stream 的
MAXLEN是近似裁剪，精确裁剪会阻塞写入，生产环境务必使用~近似模式