Redis Stream 消费组积压怎么处理：XPENDING、Claim 和排空策略

订单通知服务最近遇到一个很容易误判的现象：Redis Stream 里明明还有未处理的消息，消费者进程也没有退出，但某个消费组的 pending 数量一直在涨。这时候直接重启消费者，往往只能让新消息继续被消费，之前已经投递出去却还没确认的消息，还是会卡在原来的消费绑定关系里。

处理 Redis Stream 积压的安全顺序是：先定位 pending 的真实状态，再按空闲阈值接管超时消息，业务处理成功后确认；新消息和旧积压要分开评估，不能靠一次重启解决所有问题。

消息为什么还在 Stream 里，却不能被普通读取重新拿到

消费组通过 XREADGROUP 读取消息后，Redis 会把这条消息放进对应分组的 Pending Entries List（PEL，待确认消息列表）。只要消费者没有执行 XACK，这条消息的所属权就会一直绑定在对应消费者名下，不会因为消费者进程崩溃就自动回到待读取队列。

所以监控里同时出现“Stream 总长度正常”和“消费组 pending 持续增长”完全不冲突。前者记录的是 Stream 整个消息日志的长度，后者统计的是已经投递出去但还没完成确认的处理状态。排查的时候要把两条路径拆开：

这一步别急着扩容消费者数量。先保存一份当前的结果，尤其是 pending 总量和最大空闲时长，这两个值决定了你后续应该调整消费速度，还是先做异常消息的接管操作。

用 XPENDING 找到真正卡住的消费者

只看汇总统计不够。下面的命令会列出一小批 pending 的详情，包含消息 ID、所属消费者、投递次数和空闲时间：

XPENDING orders:events notify-group - + 20

如果返回结果里反复出现 worker-02，并且空闲时间已经超过 120000 毫秒，通常说明这个消费者已经异常退出、网络断开，或者业务逻辑卡在下游调用环节。如果投递次数很高但空闲时间很短，大概率只是当前批次处理速度偏慢，不能直接判定异常就执行接管。

建议把接管阈值设成“正常处理耗时的数倍”，而不是随便拍一个固定的几秒。比如单条通知通常 3 秒就能完成，可以先用 30 秒作为观察阈值；如果下游服务偶尔会有一分钟左右的抖动，就应该把阈值和重试策略一起同步调整。

用 XAUTOCLAIM 分批接管超时消息

确认消息确实长时间没有处理进展后，再让健康的消费者接管。Redis 6.2 及以上可以使用 XAUTOCLAIM，它会从指定消费者名下扫描符合最小空闲时间条件的消息，自动把它们转交给当前执行命令的消费者：

XAUTOCLAIM orders:events notify-group worker-01 30000 0-0 COUNT 50

这里的 30000 是判定超时的最小空闲毫秒数，COUNT 50 是本次扫描的消息数量上限。返回结果里的消息仍然需要业务代码完整执行处理逻辑，全部成功后再用 XACK orders:events notify-group 做确认。接管操作本身不代表通知已经发出，更不代表下游业务已经落库。

如果使用的 Redis 版本较老，也可以用 XPENDING 遍历找到目标消息后配合 XCLAIM 手动接管，但这种方式需要自己维护分页和消息 ID 游标，脚本在处理大积压时很容易出现重复扫描的问题。不管用哪种方式，都要限制每轮处理的消息数量，让新消息消费和旧消息排空共享明确的资源配额。

排空策略的关键不是快，而是不会重复产生异常业务结果

同一条消息可能经历原消费者处理一半超时、被新消费者接管、两个消费者的处理逻辑都接近完成的场景。Redis 只负责维护投递关系，不会帮你兜底业务侧的副作用。所以通知发送、库存变更、积分写入这类操作，必须自己实现幂等保护。

-- 伪代码：业务成功后再确认
for message in autoClaim(stream, group, consumer, 30000, 50) {
    if alreadyDone(message.id) {
        ack(message.id)
        continue
    }
    if handleBusiness(message) == nil {
        markDone(message.id)
        ack(message.id)
    } else {
        recordRetry(message.id)
    }
}

幂等键可以直接使用 Stream 自带的消息 ID，也可以用业务事件里的订单号加事件类型的组合值。处理失败时不要立即执行确认操作，否则 pending 里的对应记录会消失，但业务侧也不会留下可追溯的失败记录。

排空脚本还应该设置三道边界规则：单轮最多领取多少条消息、单条消息最多重试几次、失败消息最后流转到哪里。超过重试上限的消息可以转入独立的失败 Stream，同时保留原始消息 ID、错误原因和最后处理时间，方便后续人工复核。

上线后看三组信号，判断积压是否真的消失

把排空脚本部署到生产前，先在一组影子数据上验证“领取—处理—确认”的完整链路。上线后至少跟踪下面三组信号：

• pending 数量：连续多个采样周期稳步下降，且没有新的异常消费者持续出现。
• 空闲时间：最大值回到正常处理耗时的范围内，而不是只看统计总数变小。
• 业务重复率：幂等命中、重试、失败转存和实际成功数的统计能够对应上。

如果 pending 数量下降但失败 Stream 的消息量快速增长，说明只是把积压问题换了个位置存储；如果 pending 总量不下降反而新消息的 lag 开始增长，应该暂时调小排空批次，优先恢复正常消费能力。不要急着判定 Redis 服务已经恢复，业务侧的成功处理记录才是最终的验收标准。

常见问题

重启消费者后 pending 会自动重新投递吗？

不会。重启只会让进程重新建立连接，原有的待确认消息仍然保留在消费组的 PEL 列表中，需要通过主动接管或者明确的重试流程处理。

XAUTOCLAIM 的最小空闲时间应该设多大？

按正常单条处理耗时和下游系统最长可接受延迟来定，通常取正常耗时的数倍，同时结合投递次数做二次复核，避免接管到正在正常处理的消息。

领取消息后可以马上 XACK 吗？

不建议。只有业务副作用已经执行成功，或者确认该消息已经被幂等记录覆盖时才执行确认；否则失败消息会直接从 pending 中消失，没有回溯机会。

什么时候该增加消费者，而不是做消息接管？

如果 pending 里的消息空闲时间都不长、正常消费者仍在稳定处理，只是新消息生成速度持续高于消费速度，优先增加处理能力；如果某个消费者长时间失联，先接管它名下留下的消息。

小结

Redis Stream 积压要先区分日志长度、消费组 lag 和 pending 三个不同指标。XPENDING 用来定位异常卡住的消息，XAUTOCLAIM 或 XCLAIM 用来完成超时消息接管，XACK 只在业务完全执行成功后触发。把空闲阈值、批次上限、幂等键和失败转存逻辑一起提前设计好，整个排空过程才不会从“消息卡住”变成“业务重复执行”。