Redis 怎样自动扫描并认领空闲 Stream 消息

使用 XAUTOCLAIM 可以扫描消费组的待确认消息列表（PEL），把达到空闲阈值的消息认领给指定消费者。首次从 0-0 开始，后续传入命令返回的游标，直到返回 0-0 才结束本轮扫描；业务处理成功后再对对应 ID 执行 XACK。空闲表示消息一段时间未被重新投递或认领，不等于原消费者一定死亡，也不提供业务恰好执行一次的保证。

官方命令地址：https://redis.io/docs/latest/commands/xautoclaim/；确认语义：https://redis.io/docs/latest/commands/xack/。本文围绕已有消费组的空闲阈值、扫描游标和确认，不展开完整队列架构。下面的通知任务是原创假设场景，命令与示例代码未在本文制作过程中执行，不是运行日志。

什么时候触发回收，而不是继续读取新消息

设想通知消费者拿到任务后退出，任务已经进入 PEL，却一直没有确认。新的消费者只用 XREADGROUP 的新消息模式读取，不等于自动处理别人留下的 pending。回收任务应独立按周期运行，针对明确的 Stream、消费组和目标消费者，处理这类已投递但未确认的消息。

“自动”指应用或调度器主动调用并处理回复，不是配置一个阈值后 Redis 自己执行通知业务。调度任务还要限制单轮页数、批量大小和下游并发，不能一边大量认领，一边让新消费者的处理队列无界增长。

先定空闲阈值和目标消费者的权限范围

min-idle-time 的单位是毫秒。60000 表示至少空闲约一分钟才可认领，不是一分钟执行一次命令的调度参数。实际阈值应大于正常任务处理耗时并保留余量；若正常通知有时需要两分钟，却设成十秒，回收者可能与仍在工作的原消费者同时处理同一项。

回收账号至少需要对指定键执行所用 Stream 命令的授权；XAUTOCLAIM 会改变所有权，是写操作，不是只读诊断。读取新任务、查询 pending 和确认也有各自权限。不要为了消除权限报错直接给账号开放全部键。目标消费组应已经存在，NOGROUP 应作为配置错误处理，而非循环里不断重建。

沿返回游标扫描，空页也要继续判断

COUNT 限制的是每次尝试认领的消息数量，不保证每页都能拿满。扫描过程中，很多条目可能还没达到阈值，所以即使返回消息列表为空，也可能仍有非零游标需要继续。不要自己对最后一个消息 ID 加一，也不要使用“列表长度小于 COUNT”作为结束条件。

返回 0-0 表示这轮扫描到了末尾，不等于 PEL 清空。之前不满足空闲条件的消息，稍后可能变得可认领；到下一次调度再从头开始即可。不要在一次紧循环中遇到 0-0 后立刻无限重启。

以下是 redis-py 风格的核心函数示例，假定传入客户端使用 decode_responses=True，连接认证已在外部配置；服务端采用 Redis 7 及以上的三项回复格式。on_message 必须在业务结果已持久化或幂等判重确认后才返回 True。示例只展示一个有上限的扫描切片，不是可直接部署的完整消费者。

def recover_slice(client, on_message, cursor="0-0", max_pages=20):
    # 调用方传入上次暂停的游标；返回 0-0 后交回调度器等待下一轮。
    for _ in range(max_pages):
        reply = client.execute_command(
            "XAUTOCLAIM", "demo:notify:reclaim", "notify-group",
            "recovery-a", 60000, cursor, "COUNT", 10
        )
        next_cursor, messages, missing_ids = reply
        for message_id, field_values in messages:
            # 此回调必须处理幂等性；失败项不确认，留待后续重试。
            if on_message(message_id, field_values):
                client.execute_command(
                    "XACK", "demo:notify:reclaim", "notify-group", message_id
                )
        # missing_ids 是已无正文且从 PEL 清理的 ID，不作为业务成功计数。
        # 实际任务应将其写入独立诊断记录。
        cursor = next_cursor
        if cursor == "0-0":
            return cursor
    return cursor  # 达到页数预算，由调度器保存进度后稍后继续。

网络错误、回调异常或确认请求失败会向外抛出，由调度器退避并记录。函数没有宣称“确认失败就能回滚已提交业务”。调用方可在中断后从已保存游标继续，或从头重新扫描，但都必须允许消息重复出现；游标不是业务提交凭证。max_pages 应配置为正整数，实际回调也应有超时和资源上限。

用独立测试流观察认领和确认的区别

下面命令仅供在隔离测试实例中逐条执行，键 demo:notify:reclaim 必须尚不存在；如果已存在，换一个新测试键并同步替换所有命令，不删除或覆盖原键。示例使用明确的两个消息 ID，便于对照业务处理结果。

# 创建消费组并插入两条测试消息；不使用真实通知地址或凭据。
redis-cli XGROUP CREATE demo:notify:reclaim notify-group 0 MKSTREAM
redis-cli XADD demo:notify:reclaim 1001-0 task demo-a
redis-cli XADD demo:notify:reclaim 1002-0 task demo-b
# 让原消费者收到消息但不确认，形成 pending。
redis-cli XREADGROUP GROUP notify-group worker-a COUNT 2 STREAMS demo:notify:reclaim '>'
# 等消息空闲达到一分钟后再执行；不要在生产把阈值改成 0 来抢消息。
redis-cli XAUTOCLAIM demo:notify:reclaim notify-group recovery-a 60000 0-0 COUNT 10

预期符合阈值的消息会返回并归 recovery-a 所有。具体回复游标以执行结果为准；若仍非零，下一次调用使用它继续。没有返回正文时，先看空闲时间与回复格式，而不是立即调低阈值。

# 仅在 demo-a 的业务处理确实成功后确认第一条。
redis-cli XACK demo:notify:reclaim notify-group 1001-0
# 查看消费组剩余 pending；第二条没有成功就不要跟着确认。
redis-cli XPENDING demo:notify:reclaim notify-group

认领两条后只成功处理一条，就只确认那一条。XACK 从该消费组的 PEL 移除对应记录，不等同于删除 Stream 正文，也不证明其他系统的通知事务成功。

失败、缺失正文与 JUSTID 要分别处理

认领会重置消息的空闲计时，因此阈值内一般不会立刻被别的回收者再次认领；但原消费者不会被强行终止。若业务提交成功后、XACK 前进程退出，后续仍可能重复处理同一消息，所以应使用消息 ID 或业务唯一键实现可靠的幂等逻辑。

Redis 7.0 起，回复多出第三项，列出正文已被删除或裁剪、此次从 PEL 清理的 ID。它们不是已成功重投递的消息，不能把这部分数量算作处理成功。Redis 6.2 的回复形状不同，客户端解析必须与服务端版本匹配，不能直接把三项解包写法套过去。

JUSTID 只返回认领 ID，不提供字段内容，也不会增加相应投递计数。它仍是认领，不是无副作用的预览。需要立即处理正文的任务，优先使用普通回复；不能加了 JUSTID 后还把第二项当作 ID 与字段对的列表。

把认领、处理和确认分开记账

每轮记录键与组、目标消费者、空闲阈值、页数、认领条数、业务成功条数、确认结果、缺失正文 ID 数，以及是否到达扫描末尾。业务失败和确认异常分别统计。持续高重试的消息应转交明确的人工或故障处理流程，不通过无条件 XACK 隐藏积压。

XAUTOCLAIM 会处理从未投递的新消息吗？

它扫描已有消费组的 PEL。读取新消息仍由正常消费者入口负责。

为什么认领成功后 XPENDING 还不为零？

认领只改变所有权；完成业务后还需要 XACK，其他未确认项也仍然存在。

空闲阈值越小，恢复越快吗？

不应只看速度。过小阈值会增加与正常消费者并行处理的风险，应以实际处理时长和幂等能力设定。

自动回收的核心是三件事各有依据：空闲阈值决定资格，返回游标决定扫描进度，已持久化的业务结果决定确认。让它们分开，才能恢复 pending 而不掩盖未完成工作。