3. 消费者组机制入门

上一节我们掌握了用 XADD 写入、用 XREAD 读取的基本用法。但 XREAD 有一个明显的局限：它只记录“读到哪条消息”，不记录“哪条消息被谁处理、处理完了没有”。一旦消费者崩溃，消息就可能丢失。消费者组（Consumer Group）正是为解决这个问题而设计的。它把“读取位置”和“待确认消息”分开管理，让每条消息在被显式确认之前都处于可追踪状态。

本节围绕四个核心动作展开：用 XGROUP 创建消费者组、用 XREADGROUP 以组内消费者身份读取消息、用 XACK 确认处理完成、用 XPENDING 查看未确认消息列表（PEL，Pending Entries List）。理解这四个命令的协作关系，是掌握 Redis Streams 可靠消费的起点。

一、消费者组解决什么问题

先看一个典型场景。假设有一个订单事件流 orders，三个消费者实例同时处理订单。如果都用 XREAD 从同一位置读，每条消息会被三个实例重复处理；如果各自维护不同位置，又无法保证消息不丢。消费者组把“流”切成“组”，组内多个消费者共享一个读取进度，Redis 负责把新消息分发给组内不同的消费者，每条消息默认只投递给一个消费者。

更关键的是，消费者组为每条被投递但未确认的消息维护一条 PEL 记录，记录内容包括消息 ID、所属消费者、投递时间、投递次数。只要消费者没有调用 XACK，这条消息就一直留在 PEL 中，即使消费者进程退出，消息也不会消失，后续可以被重新认领处理。这就是“至少一次”语义的基础。

二、创建消费者组：XGROUP CREATE

创建消费者组使用 XGROUP CREATE，语法为：

XGROUP CREATE key groupname id [MKSTREAM]

其中 key 是流名称，groupname 是组名，id 指定组的起始读取位置。常用的 id 取值有三种：

• 0：从流的第一条消息开始消费，适合需要处理历史数据的场景。
• $：只消费创建组之后新写入的消息，适合只关心实时事件的场景。
• 具体消息 ID（如 1700000000000-0）：从指定位置开始，适合断点续传或补数据。

如果流还不存在，直接创建组会报错。加上 MKSTREAM 选项可以让 Redis 自动创建空流：

XGROUP CREATE orders order-group $ MKSTREAM

执行成功返回 OK。如果组已存在，会返回 BUSYGROUP Consumer Group name already exists，这在自动化脚本中需要判断处理，避免重复创建导致流程中断。

三、以组内消费者身份读取：XREADGROUP

创建组之后，消费者用 XREADGROUP 读取消息。它的语法与 XREAD 类似，但必须指定组名和消费者名：

XREADGROUP GROUP groupname consumer [COUNT n] [BLOCK ms] [NOACK] STREAMS key id

这里有两个关键参数需要区分：

• id 为 >：表示读取“从未投递给本组任何消费者的新消息”，这是正常消费新消息的用法。
• id 为具体 ID（如 0）：表示读取该消费者自己的 PEL 中未确认的消息，用于重新处理自己之前没确认完的消息。

例如，消费者 consumer-1 读取一条新消息：

XREADGROUP GROUP order-group consumer-1 COUNT 1 STREAMS orders >

返回结果中会包含消息 ID 和字段内容。此时这条消息已经被投递给 consumer-1，并自动进入 PEL。注意：XREADGROUP 的读取是“投递即入 PEL”，不需要额外操作。如果加上 NOACK 选项，消息不会进入 PEL，相当于放弃可靠投递，适合允许丢失的日志类场景。

四、ACK 确认：XACK

消费者处理完消息后，必须调用 XACK 把消息从 PEL 中移除：

XACK orders order-group 1700000000000-0

返回值是成功确认的消息数量。如果返回 1，说明该消息确实在 PEL 中并被移除；返回 0 通常意味着消息 ID 不存在、已被确认过，或者根本不属于这个组。生产环境中应检查返回值，避免“以为确认了其实没确认”的隐性 bug。

这里有一个重要的工程原则：先处理业务，再 ACK。如果先 ACK 再处理，处理过程中崩溃就会丢消息；如果处理成功但 ACK 失败，消息会留在 PEL 中被重复处理，因此业务逻辑需要具备幂等性。这是“至少一次”语义下的标准做法。

五、查看 PEL：XPENDING

要了解哪些消息还没被确认，使用 XPENDING。它有两种常用形式。第一种是概要形式：

XPENDING orders order-group

返回四部分信息：未确认消息总数、最小消息 ID、最大消息 ID、以及按消费者分组的未确认数量。例如返回 3、1700000000000-0、1700000000002-0、consumer-1: 2, consumer-2: 1，说明当前有 3 条消息待确认，其中 consumer-1 积压 2 条，consumer-2 积压 1 条。

第二种是详细形式，可以查看具体消息和投递次数：

XPENDING orders order-group - + 10

返回每条消息的 ID、所属消费者、空闲时间（毫秒）和投递次数。空闲时间过长或投递次数过多的消息，往往意味着消费者异常或处理逻辑卡住，是排障的重要线索。

六、四个命令的协作关系

把上面的过程串起来看：XGROUP CREATE 先建立组和读取起点；XREADGROUP 以组内消费者身份拉取消息，消息被投递的同时自动写入 PEL；消费者处理业务；处理成功后用 XACK 把消息从 PEL 移除；XPENDING 则随时可以查看 PEL 中还有哪些消息没被确认、分别属于谁、积压了多久。

这四者构成一个闭环：创建组定义消费边界，读取投递产生待确认记录，ACK 消除待确认记录，XPENDING 暴露待确认记录。缺少任何一环，可靠消费都不完整。例如只读不 ACK，PEL 会无限增长，内存和排障成本都会上升；只 ACK 不查看 PEL，则无法发现消费者异常。

七、一个可复现的示例流程

下面用一组命令演示完整流程，假设流 orders 已有若干消息：

# 1. 创建消费者组，从最新消息开始消费
XGROUP CREATE orders order-group $ MKSTREAM

# 2. consumer-1 读取一条新消息
XREADGROUP GROUP order-group consumer-1 COUNT 1 STREAMS orders >

# 3. 查看 PEL 概要，确认消息已进入待确认列表
XPENDING orders order-group

# 4. 处理业务后确认消息
XACK orders order-group 1700000000000-0

# 5. 再次查看 PEL，确认消息已被移除
XPENDING orders order-group

执行第 3 步时，如果看到未确认数量为 1、消费者为 consumer-1，说明投递成功；执行第 5 步时，如果未确认数量回到 0，说明 ACK 生效。如果第 5 步仍然显示 1，需要检查 ACK 的消息 ID 是否与投递时返回的 ID 完全一致——ID 写错是初学者最常见的失误。