Stream 消费者组与消息队列

Stream 是 Redis 5.0 引入、唯一自带消费者组的类型，可以当轻量级消息队列使用：消息有序、支持阻塞读取、支持 ACK 确认与宕机找回。它不是 Kafka 的替代品，但在"量不大、想少引入一个组件"的场景非常合适。本文以 Redis 7 为基准，把生产消费与运维细节讲全。

一、XADD 与消息 ID

消息 ID 的格式是 毫秒时间戳-序号，XADD 用 * 自动生成并保证严格递增；手动指定 ID 必须大于流内最大 ID，否则报错。XADD 可以顺带裁剪：MAXLEN ~ 1000 中的波浪号表示允许略微超标，换取裁剪性能；MINID 按 ID 裁剪；NOMKSTREAM 表示流不存在时不自动创建：

127.0.0.1:6379> XADD orders MAXLEN ~ 1000 * uid 1001 item p30 amount 199
"1698244800000-0"
127.0.0.1:6379> XLEN orders
(integer) 1
127.0.0.1:6379> XRANGE orders - +
1) 1) "1698244800000-0"
   2) 1) "uid"
      2) "1001"
      3) "item"
      4) "p30"
      5) "amount"
      6) "199"

XRANGE/XREVRANGE 按 ID 范围读取，- 与 + 表示头尾，ID 可以省略序号部分只写毫秒值。按时间段拉取历史消息就是靠 XRANGE 加时间戳范围实现。

手动指定 ID 主要用在两处：数据导入时保持原有时间序，以及业务要求"同一毫秒内严格按提交顺序"时自己控制序号。XADD 的字段是 field-value 对，同一字段写两次以后写的值为准。消息体要保持扁平，建议只放业务 ID 与少量路由字段，详情另查存储，避免流变成大 value 的聚集地。MAXLEN 的裁剪发生在写入路径上，取值要按消费能力与保留需求权衡：太小可能裁掉还没消费的消息，太大会内存膨胀，上线前用 XLEN 与 XINFO STREAM 观察增长速度再定。

二、消费者组：核心概念

消费者组由 XGROUP 创建，维护两个字段：last-delivered-id（组内已投递到的位置）和每个消费者的 PEL（pending entries list，已投递未确认列表）。

# 从 0 开始消费历史消息；MKSTREAM 在流不存在时自动创建
XGROUP CREATE orders grp 0 MKSTREAM
# 只消费创建之后的新消息
XGROUP CREATE orders grp2 $ MKSTREAM

消费者用 XREADGROUP 拉取：

XREADGROUP GROUP grp consumer1 COUNT 10 BLOCK 5000 STREAMS orders >

> 表示只取从未投递过的新消息；把 > 换成 0 则读取自己 PEL 里未确认的消息，用于恢复现场。同一个组内多个消费者是负载均衡关系，每条消息只投给一个消费者；不同组之间是广播关系，各自维护进度。NOACK 选项跳过 PEL 直接消费，适合允许少量丢失的场景，能显著减少 PEL 维护开销。

不带组的 XREAD 也能读流（支持 BLOCK 阻塞、按 ID 游标），适合"每个消费者都要全量消息"的简单广播，但它没有投递记录，宕机后不知道读到哪，游标要消费方自己持久化。XINFO GROUPS 里每个组都有 pending 与 consumers 计数；XGROUP SETID 可以手动改组的投递位置，主要用于故障后的进度修复，正常流程不要动它。消费者的名字是客户端在 XREADGROUP 时自报的，首次出现即自动注册，不需要预先创建；XINFO CONSUMERS 能看到每个消费者的空闲时间与 PEL 数量，排查"谁没消费"从这里入手。

三、ACK 与宕机找回

处理成功后 XACK 确认：XACK orders grp <message-id>。消费者崩溃时消息留在 PEL 里，找回流程是三步：

# 1. 查看 PEL，可按空闲时间过滤（IDLE 单位毫秒，6.2+）
XPENDING orders grp IDLE 60000 - + 10
# 2/3. 把空闲超 60 秒的消息转移给 consumer2（XAUTOCLAIM 为 6.2+ 命令）
XAUTOCLAIM orders grp consumer2 60000 0-0 COUNT 10

XPENDING 负责观察，XCLAIM/XAUTOCLAIM 负责转移，XAUTOCLAIM 会自动扫描 PEL 并游标式返回，比手动 XCLAIM 省心，Redis 7 起返回值还会附带已被 XDEL 的消息 ID 方便清理。这套"PEL 加空闲认领"就是 Stream 的重试机制，但重试次数要业务自己维护——认领时读消息里的重试计数字段，超过阈值就转写死信流，Stream 没有内建死信队列。

精准干预用 XCLAIM：先 XPENDING 拿到具体 ID，确认对方确实失联后 XCLAIM orders grp consumer2 60000 <id> 转移。它与 XAUTOCLAIM 的区别在于：XCLAIM 精确指定消息 ID，XAUTOCLAIM 自动扫描、批量转移、顺带清理已删除消息的残留。日常巡检统一用 XAUTOCLAIM，XCLAIM 留给需要人工确认的定向接管。认领后的消息要重新走完整处理流程，包括业务校验。XAUTOCLAIM 的 min-idle-time 设太小会抢走慢消费者正在处理的消息，设太大又拉长故障恢复时间，一般取"正常处理耗时的 5 到 10 倍"，再结合告警数据调整。

四、运维细节

三个必须知道的点。第一，XDEL 只做删除标记，内存要等所在的 listpack 节点整块清空才真正释放，控制内存靠 MAXLEN/MINID/XTRIM 保持流有界，而不是指望 XDEL 省内存。第二，删除消费者用 XGROUP DELCONSUMER 前必须确认其 PEL 已清空——该命令会把它未确认的消息从 PEL 一并移除，这些消息之后既不能 ACK 也不能被认领，等于对组内"消失"。第三，用 XINFO GROUPS、XINFO CONSUMERS、XINFO STREAM 观察组、消费者与流的健康度，重点看 PEL 长度与消费者空闲时间，配上告警。

积压的应急处理有个顺序：先分清是消费变慢还是消费停了——停了先恢复消费者实例；慢了就在组内多加几个消费者横向扩（Stream 的组内扩容不用改流结构）；再不行检查单条消息处理是否太重，把重处理挪成异步。最后手段是新建组从最新位置开始消费、旧消息走离线补偿，这是数据取舍，必须业务拍板而不是运维单方面决定。最后，Stream 是 at-least-once 语义，重复投递不可避免，消费端必须幂等（见同系列幂等设计文章的通用做法）。

常见坑 / 最佳实践

1. 监控 PEL 长度并配置 XAUTOCLAIM 巡检。 忘记 XACK 的消息会永远躺在 PEL，故障恢复依赖定期的空闲认领任务。
2. 流必须有界。 XADD 时带 MAXLEN ~ 或定期 XTRIM，靠 XDEL 控制内存是无效的；历史消息的保留时长与业务对齐。
3. XGROUP CREATE 的起始 ID 语义要明确。 $ 会"失忆"创建前的消息，服务重建组时用 0 还是 $ 要写成文档，避免静默丢消息。
4. DELCONSUMER 前先清空 PEL。 它会连带删除未确认消息的投递记录，这些消息再也无法 ACK 或认领。
5. 消费者逻辑必须幂等。 at-least-once 加空闲认领意味着必然重复投递，用业务唯一键去重是底线。

小结

• 消息 ID 是 毫秒-序号 严格递增，XADD 支持 MAXLEN ~ 顺带裁剪，历史读取走 XRANGE。
• 消费者组维护 last-delivered-id 与 PEL：组内负载均衡、组间广播，> 取新消息、0 取自己未确认消息。
• 宕机找回三步走：XPENDING 观察、XAUTOCLAIM 转移、超限转死信流，重试计数由业务维护。
• XDEL 不释放内存，靠 MAXLEN/XTRIM 保流有界；DELCONSUMER 会连带清 PEL，操作前必须确认。