编程 Redis 8.4 XREADGROUP CLAIM：一条命令同时取空闲 pending 和新增条目

2026-10-19 21:04:19

Redis 8.4 XREADGROUP CLAIM：一条命令同时取空闲 pending 和新增条目

问题：pending 会卡住

Streams 里消息投递给消费组的某个消费者后进入 PEL（pending entries list），直到被 XACK 或删除才消失。长时间 pending 通常意味着出了问题：消费者在处理中崩溃、在 ack 之前挂掉；消息本身有问题（死锁、极慢）；或者通信异常。应用一般假设「消费后 X 秒内必须 ack」，超过这个时间就是空闲 pending，应该重新投递。

所以恢复逻辑必须同时做两件事：扫描 pending 列表，把空闲的认领回来；以及继续处理新进入的条目。8.4 之前，这两件事得自己拼。

旧循环：两段式

cursor = "0-0"
while True:
    # 1) 认领空闲 pending，游标要自己维护
    cursor, msgs, _ = redis.xautoclaim(
        name, group, consumer,
        min_idle_time=30000, start_id=cursor, count=100)
    for msg_id, fields in msgs:
        handle(msg_id, fields)
        redis.xack(name, group, msg_id)

    # 2) 再取新条目
    resp = redis.xreadgroup(group, consumer, {name: ">"}, count=100, block=5000)
    for _, entries in resp or []:
        for msg_id, fields in entries:
            handle(msg_id, fields)
            redis.xack(name, group, msg_id)

两个分支的批量大小、阻塞时长、错误处理各写一遍，XAUTOCLAIM 的游标还得在循环里记住上一次的返回值，否则每轮都从头扫。

CLAIM：把认领塞进 XREADGROUP

8.4 给 XREADGROUP 加了 CLAIM min-idle-time 选项：一次调用里既认领空闲 pending，也返回新增条目。

XREADGROUP GROUP g c COUNT 100 BLOCK 5000 CLAIM 30000 STREAMS s >

几个要点：

CLAIM 写在 STREAMS 之前，参数是毫秒级的 min idle time，作用于「本消费组内、空闲时间超过该值」的 pending 条目。
> 只表示新条目。要读消费者自己的 PEL，传 0 或具体 ID。带 CLAIM 时，一次调用把两类都返回。
部署前先对着命令文档核对参数顺序。

min-idle-time 怎么定

它是「多久没 ack 就认定这个消费者不行了」的判定阈值，所以下限必须大于正常处理耗时。

下限：p99 处理耗时加 ack 往返。设得比正常处理还短，就会把正在处理的消息抢走，白白制造重复处理。
上限：能容忍的最长恢复延迟。设得太大，消费者崩了之后消息要躺很久才被重投。
跟重试预算挂钩：若业务允许一条消息重试 N 次，第 k 次重投的等待量级是 k × min-idle-time，据此反推阈值。同时注意每次重投 delivery count 都会增长，可以用它识别毒消息，超过阈值直接进死信，不要无限重投。

READ 与 CLAIM 的取舍

min-idle-time 的调整是双向的：太小会抢正在处理的消息，太大会拖慢恢复。除此之外还要决定 CLAIM 该不该放在主循环里：

主循环带 CLAIM：代码路径唯一，一个 while 覆盖新增与恢复，消费者数量少时最省事。
主循环只读 >，另起低频任务跑 CLAIM（或继续用 XAUTOCLAIM）做恢复：主循环的阻塞时间稳定，恢复频率可控，出问题时能单独关掉。

pending 上限与 XPENDING 监控

COUNT 只决定每轮取多少，PEL 本身没有自动上限。消费者长期落后时 pending 会一直涨，所以至少要盯三个数：

XPENDING s g          # 总数、最小/最大 ID、各消费者条数
XPENDING s g - + 10   # 明细，含空闲时长与 delivery count

总量、最老一条的空闲时长、按消费者分布。某个消费者的 pending 只增不减，基本就是它挂了或卡住了。

幂等仍然要自己做

CLAIM 会把消息重新投递，包括「其实已经处理完、只是没来得及 ack」的那种。重复消费是设计内行为，去重得靠业务侧：消息 ID 落库做唯一键，或者把处理逻辑写成幂等的。XACK 紧跟处理成功之后，尽量缩短这个窗口。

已知限制

8.4 中，原子槽位迁移（atomic slot migration）期间 FT.* / TS.* 可能返回部分结果或重复数据；Streams 命令不在这个列表里。