Redis XAUTOCLAIM 怎样只取消息 ID 做恢复

在 Redis Streams 的消费组里，消费者崩溃后消息会留在 Pending Entries List（PEL）中。恢复程序如果只是想“接管哪些消息”，并不需要在扫描阶段把每条消息的字段都拉回来：给 XAUTOCLAIM 加上 JUSTID，第二个返回元素就变成成功接管的消息 ID 列表，应用随后按 ID 读取、处理，最后再用 XACK 确认。

官方地址：https://redis.io/docs/latest/commands/xautoclaim/

核心顺序是“只用 JUSTID 接管 → 按 ID 读取并处理 → 成功后 XACK”。JUSTID 只减少接管扫描的返回体，不代表消息已经处理完成，也不会替你删除 PEL。

项目目标：做一个只搬运 ID 的恢复器

假设订单消费组里有一个消费者下线，若干消息已经 pending 超过 60 秒。恢复器需要把这些消息转交给 repair-worker，但扫描阶段只拿到消息 ID。这样做适合消息字段很大、恢复任务只想先建立工作清单，或者应用已经有按 ID 读取的处理函数的场景。

这个小项目只覆盖三个动作：按最小空闲时间扫描 PEL、用游标继续扫描、处理成功后确认。它不讨论 Stream 裁剪策略，也不把“成功接管”误写成“业务成功”。

环境准备：先准备一组可恢复的 pending 消息

XAUTOCLAIM 自 Redis Open Source 6.2.0 起可用。下面用 Redis CLI 建立一个 Stream 和消费组，并让 worker-a 先读取消息但暂不确认，从而形成待恢复数据。

# 创建一条订单事件；* 让 Redis 生成 Stream ID
redis-cli XADD orders * order_id 1001 status paid

# 创建消费组；$ 表示只接收组创建之后的新消息
redis-cli XGROUP CREATE orders order-workers $ MKSTREAM

# 模拟 worker-a 读取一条消息但暂不 XACK，使它留在 PEL
redis-cli XREADGROUP GROUP order-workers worker-a COUNT 1 STREAMS orders >

# 查看消费组的 pending 概况
redis-cli XPENDING orders order-workers

在真实环境里，不需要为了演示专门延迟消息；只要某条消息已经被 XREADGROUP 投递给消费者而没有 XACK，它就会出现在 PEL。恢复器用毫秒值判断空闲时间，例如 60000 表示至少空闲 60 秒。

核心调用：JUSTID 到底改变了什么

命令格式如下：

# 从 0-0 开始扫描，接管空闲超过 60 秒的消息，最多尝试 25 条，只返回 ID
redis-cli XAUTOCLAIM orders order-workers repair-worker 60000 0-0 COUNT 25 JUSTID

普通 XAUTOCLAIM 返回三部分：下一次扫描用的游标、带字段的成功接管消息、以及已经不在 Stream 中但仍在 PEL 里的 ID。加入 JUSTID 后，第二部分变成 ID 数组，并且被接管消息的尝试次数不会递增。Redis 文档还说明，COUNT 是每次尝试扫描的上限，真正成功接管的数量可能更少。

因此，下面两件事要分开理解：

核心代码：先接管 ID，再按 ID 读取

下面的 Python 示例使用 redis-py 的底层 execute_command，可以直接看到 Redis 命令和返回结构。实际项目也可以换成客户端对 XAUTOCLAIM JUSTID 的高层封装。

import redis

STREAM = "orders"
GROUP = "order-workers"
CONSUMER = "repair-worker"
MIN_IDLE_MS = 60_000

client = redis.Redis(host="localhost", port=6379, decode_responses=True)

def recover_once(start_id: str = "0-0", count: int = 25) -> tuple[str, list[str], list[str]]:
    # JUSTID 只返回消息 ID，避免恢复扫描阶段搬运完整字段体。
    reply = client.execute_command(
        "XAUTOCLAIM", STREAM, GROUP, CONSUMER,
        MIN_IDLE_MS, start_id, "COUNT", count, "JUSTID"
    )
    # Redis 6.2 的返回重点是游标和 ID；Redis 7.0+ 还会返回被清理的失效 ID。
    next_id = reply[0]
    claimed_ids = list(reply[1])
    deleted_ids = list(reply[2]) if len(reply) > 2 else []
    return next_id, claimed_ids, deleted_ids

def process_by_id(message_id: str) -> bool:
    # 只用 ID 读取消息；读取不到时说明 Stream 可能已经裁剪或删除。
    rows = client.xrange(STREAM, min=message_id, max=message_id)
    if not rows:
        return False
    actual_id, fields = rows[0]
    # 这里替换为幂等的订单处理逻辑，成功后才允许 XACK。
    print(f"处理 {actual_id}: {fields}")
    return True

def recover_all() -> None:
    # 游标不是业务消息 ID，必须使用 XAUTOCLAIM 返回的值继续扫描。
    start_id = "0-0"
    while True:
        next_id, claimed_ids, deleted_ids = recover_once(start_id)
        for message_id in claimed_ids:
            # 处理失败不确认，让后续恢复周期仍有机会再次接管。
            if process_by_id(message_id):
                client.xack(STREAM, GROUP, message_id)
        # 被 Redis 清理的 ID 已不具备业务处理对象，只记录后跳过。
        for message_id in deleted_ids:
            print(f"跳过已不存在的消息: {message_id}")
        if next_id == "0-0":
            break
        start_id = next_id

if __name__ == "__main__":
    recover_all()

这里的关键不是把消息体永远不读，而是把“扫描和搬运所有权”与“按 ID 取得业务数据”拆开。若消息已经被裁剪，按 ID 读取会得到空结果；这时不能因为接管成功就直接 XACK，应该记录并按业务策略处理。

循环边界：0-0、空数组和 COUNT 不要混在一起

恢复器最容易出错的地方是把三个信号当成一个信号：

1. claimed_ids 为空：本轮没有成功接管的消息，不等于 PEL 已经扫描完；可能只是当前窗口没有达到最小空闲时间。
2. next_id 为 0-0：Redis 表示这次扫描已经到达 PEL 尾部。若任务是持续恢复，下一轮定时任务仍可从 0-0 重新开始，因为新的消息可能在之后变得足够空闲。
3. deleted_ids 非空：Redis 发现 PEL 中的 ID 已经不在 Stream 里，并把它从对应 PEL 中清理。它没有字段可供业务处理，不应拿去 XACK 当成成功业务。

COUNT 25 也不是“必定返回 25 条”。Redis 会按 PEL 扫描，并过滤空闲时间不满足条件的消息，所以成功接管数可能少于 COUNT。恢复器应以实际返回 ID 为准，而不是根据 COUNT 预分配成功数量。

部署集成：把恢复器放进定时任务

恢复器可以作为常驻协程，也可以每隔一段时间触发一次。无论哪种方式，都建议为它使用独立 consumer 名称，并记录起始游标、接管数量、按 ID 读取为空的数量、处理失败数量以及 XACK 返回值。

# 每分钟运行一次恢复器；输出交给日志系统，不把凭据写进命令行
* * * * * /usr/bin/python3 /opt/order-repair/recover.py >> /var/log/order-repair.log 2>&1

# 观察消费组的 pending 概况；排查时只看统计，不把消息内容写入日志
redis-cli XPENDING orders order-workers

如果多个恢复实例同时运行，Redis 的接管动作会重置消息空闲时间，同一时刻通常只有一个消费者能成功接管同一条消息。但业务处理仍应保持幂等，因为网络重试、进程中断和业务侧失败都可能带来再次投递。

验收：用命令确认“接管”和“确认”是两步

可以用下面的顺序做最小验收：

1. 用 XPENDING orders order-workers 看到待处理消息。
2. 运行一次带 JUSTID 的 XAUTOCLAIM，确认返回的是游标和 ID，而不是字段键值对。
3. 让恢复器按 ID 读取并执行幂等处理。
4. 对处理成功的 ID 调用 XACK orders order-workers message-id。
5. 再次查看 XPENDING，并以 XACK 返回的整数判断实际从 PEL 移除了多少条。

如果只做了第三步却没有第四步，消息仍会留在 PEL；如果先 XACK 再处理，处理失败时就失去了自然重试的入口。这个顺序比是否使用某个客户端封装更重要。

常见问题

JUSTID 会不会把消息内容永久删除？

不会。它只改变 XAUTOCLAIM 的返回形式，并转移消息所有权；Stream 条目本身仍由 Stream 的保留、删除或裁剪策略决定。

拿到消息 ID 后为什么还要再次读取？

因为 JUSTID 的目标就是让接管扫描阶段只搬运轻量 ID。业务处理需要字段时，再按 ID 读取可以把恢复扫描与处理阶段拆开。

返回 0-0 后还要不要下一次再扫？

持续恢复任务要在下一轮从 0-0 重新开始。0-0 表示本次扫描到达尾部，不是永久没有新 pending 消息。

XACK 返回 0 是什么情况？

通常说明该 ID 已不在这个消费组的 PEL 中，可能已被其他流程确认或清理。它不是业务处理成功数，应结合日志和 XPENDING 判断。

小结

Redis XAUTOCLAIM 的 JUSTID 适合把恢复流程拆成轻量的“接管 ID”阶段。实现时记住四点：用返回游标继续扫描；以实际 ID 数量而不是 COUNT 判断结果；按 ID 读取失败时不要盲目确认；业务成功后再调用 XACK。这样既能减少恢复扫描的返回体，也保留消费组对未完成消息的追踪能力。