Streams 消费者组积压怎么处理：认领、重试与裁剪

Redis Streams 消费者组出现“积压”时，我通常先把它拆成两类：lag 是还没有投递给组内任何消费者的消息，pending 是已经投递、却还没有执行 XACK 的消息。前者多半要提升消费吞吐，后者才需要检查失联消费者、超时任务和认领重试。两类指标混在一起，最容易出现一边盲目扩容、一边把真正卡住的消息留在 PEL 里。

官方文档：https://redis.io/docs/latest/develop/data-types/streams/

处理顺序可以记成一句话：先用 XINFO 与 XPENDING 定位积压类型，再用 XAUTOCLAIM 分批接管超时消息；业务成功后 XACK，超过重试上限进入死信，最后才按所有消费者组的确认边界裁剪 Stream。

先分清未投递 lag 与未确认 pending

XINFO GROUPS 同时提供组内消费者数、PEL 长度、last-delivered-id、entries-read 和 lag。其中 pending 是这个组已经投递但未确认的数量；lag 是仍等待投递给该组的数量。Redis 7.0 才加入 entries-read 与 lag 字段，而且在任意位置创建组或中间消息被删除、裁剪时，lag 可能暂时返回空值，因此监控代码要允许它不可用。

我会先执行下面三组只读命令，把“组没追上”和“消息处理后没确认”分开：

# 查看消费者组整体：重点观察 lag、pending 和消费者数量
redis-cli XINFO GROUPS orders:events

# 查看 workers 组内各消费者的 pending、idle 与 inactive
redis-cli XINFO CONSUMERS orders:events workers

# 查看 PEL 摘要：总数、最小/最大 ID 以及各消费者分布
redis-cli XPENDING orders:events workers

定位真正需要接管的消息

XPENDING 的扩展形式可以返回消息 ID、当前所有者、空闲毫秒数和投递次数。不要看到 pending 就全部抢走；正在处理一个耗时任务的健康消费者，也会暂时留下 pending。min-idle-time 应大于正常处理耗时的高分位值，并额外留出网络抖动、下游重试和进程暂停的余量。

# 只查看空闲至少 60 秒的 pending，最多取 20 条用于判断
redis-cli XPENDING orders:events workers IDLE 60000 - + 20

# 若只怀疑某个消费者，可在末尾加消费者名缩小范围
redis-cli XPENDING orders:events workers IDLE 60000 - + 20 worker-3

XINFO CONSUMERS 里的 idle 与 inactive 也不能混用。当前文档中，idle 表示距离最近一次尝试交互的时间，inactive 表示距离最近一次成功交互的时间；Redis 7.2 以前的 idle 语义不同。如果监控同时覆盖多个 Redis 版本，先确认字段是否存在及其版本含义。

用 XAUTOCLAIM 分批接管失联消息

Redis 6.2 起提供 XAUTOCLAIM。它把“扫描 PEL + 判断空闲时间 + 改变所有者”组合成一个带游标的命令，适合恢复消费者宕机后留下的消息。相比手动 XPENDING 再逐条 XCLAIM，它更容易持续小批量扫描，也减少了客户端拼装认领列表的工作。

# rescue-1 接管空闲至少 60 秒的消息，从 PEL 起点扫描，单批最多认领 50 条
redis-cli XAUTOCLAIM orders:events workers rescue-1 60000 0-0 COUNT 50

返回值的第一项是下一次扫描要使用的游标，第二项是本次成功认领的消息。应用应把下一游标传回下一次 XAUTOCLAIM，直到游标返回 0-0，而不是每次都从头扫。COUNT 是尝试认领数量的上限，不保证每次正好返回这么多条；空闲时间不满足条件的项会被跳过。

JUSTID 只返回 ID，并且不会增加重试计数，适合先做轻量盘点，不适合依赖投递次数做重试上限的主处理路径。认领只改变 PEL 所有者，不代表业务执行成功，也不会自动确认消息。

重试必须有上限，处理器必须幂等

消费者组提供的是至少一次处理语义：消费者可能在业务写入成功后、执行 XACK 前崩溃，消息随后被其他消费者认领并再次执行。对我来说，认领功能是否好用，关键不在命令本身，而在业务副作用能否按消息 ID、订单号或幂等键去重。

可以按投递次数把消息分为三个区间：

• 首次或少量重试：正常执行幂等处理，成功后确认。
• 接近上限：降低并发，记录完整错误原因，避免故障下游被持续放大。
• 超过上限：写入独立死信 Stream，再从原消费者组确认，交给人工或离线任务处理。

# 业务处理成功后再确认；返回值表示真正从 PEL 移除的 ID 数量
redis-cli XACK orders:events workers 1730781000000-0

# 超过重试上限时写入死信 Stream，并保留原消息 ID 与失败原因
redis-cli XADD orders:dead '*' source_id 1730781000000-0 reason 'downstream_timeout'

# 确认死信已经可靠写入后，再确认原消费者组中的消息
redis-cli XACK orders:events workers 1730781000000-0

上面两条死信命令用于说明边界，并不天然覆盖外部数据库事务。生产实现需要让“死信去重”和“原消息确认”具备可恢复性；如果两者都在同一个 Redis 中，可以评估事务、Lua 或 Redis Functions，但仍要为客户端超时后的未知结果准备幂等键。最危险的顺序是先 XACK 再执行业务，因为后续失败时消息已经从 PEL 消失。

认领不能和新消息消费抢光资源

恢复程序如果一次接管大量旧消息，可能把数据库连接、HTTP 配额或 CPU 全部占满，反而让 lag 继续增长。我更倾向于把“读取新消息”和“恢复超时 pending”设成两条有界通道：两边都有固定批次、并发上限和超时，恢复通道只拿一部分容量。

# 正常消费者只读取尚未投递的新消息；COUNT 限制单次批量
redis-cli XREADGROUP GROUP workers worker-1 COUNT 50 BLOCK 2000 STREAMS orders:events '>'

# 恢复消费者另设更小批量，避免旧消息重试压垮下游
redis-cli XAUTOCLAIM orders:events workers rescue-1 60000 0-0 COUNT 20

扩容消费者主要解决 lag，并不会自动处理其他消费者名下的 pending。相反，只运行认领程序也不会提升新消息吞吐。监控应同时看组级 lag 趋势、PEL 数量、最老 pending 的空闲时间、投递次数分布和业务处理错误率。

裁剪前先理解 Stream 与 PEL 是两层状态

XACK 移除的是某个消费者组 PEL 中的引用，不会从 Stream 主体删除消息。XTRIM 删除的是 Stream 中较旧的 entry；如果裁剪早于消费恢复，PEL 可能还保留 ID，但对应 payload 已经不在 Stream 中。默认把长度裁短并不等于“只删已经处理完的消息”。

Redis 8.2 为 XTRIM 增加了消费者组引用策略：

• KEEPREF 是默认行为：裁剪 Stream entry，但保留现有 PEL 引用。
• DELREF 同时删除所有消费者组中的相关 PEL 引用，属于更强的清理动作。
• ACKED 只裁剪已被所有消费者组确认的 entry，更适合多组共享同一 Stream 的保留策略。

# Redis 8.2+：近似保留约 10 万条，但只裁剪所有组都已确认的消息
redis-cli XTRIM orders:events MAXLEN '~' 100000 LIMIT 5000 ACKED

# Redis 6.2+：按最小 ID 近似裁剪；阈值必须来自业务保留边界评估
redis-cli XTRIM orders:events MINID '~' 1730000000000-0 LIMIT 5000

~ 是近似裁剪，通常比精确的 = 更高效，但可能暂时保留超过阈值的 entry。LIMIT 控制本次检查量，适合把清理工作摊到多次调用中。若运行版本早于 8.2，就没有 ACKED 选项；这时不能照搬新命令，应根据所有消费者组的处理进度、最老 pending ID 和业务保留时长制定保守的 MINID 或 MAXLEN，并给故障恢复留出窗口。

和旧的 XPENDING + XCLAIM 方案怎么选

我会采用的积压处理清单

1. 用 XINFO GROUPS 记录 lag、pending、消费者数和 last-delivered-id。
2. 用 XINFO CONSUMERS 与扩展 XPENDING 找到高 idle、高投递次数的消息。
3. 把 min-idle-time 设在正常处理耗时高分位以上，避免误抢。
4. 用 XAUTOCLAIM 小批量扫描，并持久化 next cursor 直到一轮结束。
5. 业务处理按消息 ID 或业务键幂等；成功后再 XACK。
6. 设置有限重试和死信 Stream，避免毒消息无限循环。
7. 正常消费与恢复消费分别限流，持续观察 lag 是否停止增长。
8. 确认所有消费者组的保留边界后再裁剪；Redis 8.2+ 优先评估 ACKED。

这套方法的核心不是“把 pending 清零”，而是保持三个边界一致：消息何时可以被其他消费者接管、失败多少次后不再自动重试、历史 payload 到什么位置才允许删除。只要这三个判断能够从指标和业务状态中恢复，消费者重启、网络超时和重复投递就不会把积压处理变成一次高风险清库操作。

常见问题

pending 很多，能不能直接 XACK 批量清掉？ 只有能够证明业务已经成功完成时才可以。直接确认只会让 Redis 停止跟踪这些消息，不能补做业务处理。

XAUTOCLAIM 会保证消息只执行一次吗？ 不会。它改变所有者并允许重新处理，消费者组整体仍是至少一次语义，业务侧必须幂等。

lag 为 0 就表示没有积压吗？ 不一定。lag 为 0 只说明没有等待首次投递的消息，PEL 中仍可能有大量未确认消息。

为什么 MAXLEN 已设置，Stream 仍稍微超过阈值？ 使用近似裁剪 ~ 时允许暂时保留更多 entry；使用 ACKED 时，仍被某个组引用的消息也可能阻止达到目标长度。

官方参考

• Redis Streams 数据类型与消费者组
• XINFO GROUPS
• XPENDING
• XAUTOCLAIM
• XACK
• XTRIM