Redis Stream的消费者组与消息确认机制

核心概念解析

Redis Stream是一种强大的数据结构，用于处理消息队列。要高效且安全地消费消息，必须理解其两大核心机制：消费者组和消息确认。

消费者组 允许你将消息流划分给多个消费者。每个消费者组独立维护一个“游标”，记录该组已消费到的位置。组内的每个消费者都能接收到消息，但同一条消息在组内只会被一个消费者处理。这实现了负载均衡。

消息确认（ACK） 是确保消息不被重复处理的关键。当消费者成功处理一条消息后，必须向服务器发送一个“确认”信号。服务器随后会将该消息标记为“已处理”。如果消费者在确认前崩溃，服务器会将消息重新分配给同组的其他消费者，避免消息丢失。

第一步：创建消费者组与消费者

创建消费者组 的前提是流（Stream）已存在。

1. 添加 测试数据到名为 mystream 的流中，键为 sensor_id 和 value。

   XADD mystream * sensor_id 1 value 25.6
   XADD mystream * sensor_id 2 value 26.1
   XADD mystream * sensor_id 3 value 24.8

2. 创建 消费者组，组名为 mygroup，指定从流的最后一条消息开始消费（`$`）。 ```bash XGROUP CREATE mystream mygroup $ MKSTREAM

• MKSTREAM 表示如果流 mystream 不存在，则自动创建它。

3. 查看 当前消费者组信息。

   XINFO GROUPS mystream

   此时输出中 last-delivered-id 应为 0-0，表示该组尚未消费任何消息。

4. 创建 组内的消费者 consumer1 和 consumer2。消费者会在第一次读取消息时自动创建，但也可预先创建。

   XGROUP CREATECONSUMER mystream mygroup consumer1
   XGROUP CREATECONSUMER mystream mygroup consumer2

5. 读取 消息。使用 XREADGROUP GROUP 命令，指定组和消费者。

   XREADGROUP GROUP mygroup consumer1 COUNT 1 BLOCK 2000 STREAMS mystream >

• GROUP mygroup consumer1：以 mygroup 组的 consumer1 身份读取。
• COUNT 1：读取1条消息。
• BLOCK 2000：如果无消息，阻塞等待2秒（毫秒）。
• STREAMS mystream >：从流 mystream 中，读取仅限该消费者组未消费过的新消息（> 特殊ID）。

6. 执行 第5步命令后，consumer1 将接收到流中的第一条消息。该消息同时进入该消费者组的“待处理（PENDING）”列表。

第二步：查看与管理待处理（PENDING）消息

消息被 XREADGROUP 读取后，在确认（ACK）之前，它一直处于“待处理”状态。消费者必须维护一个本地未确认消息列表，或通过Redis命令查询。

1. 查看 mygroup 组中所有待处理消息的概要信息。

   XPENDING mystream mygroup

   输出会显示：最小待处理消息ID、最大待处理消息ID、待处理消息总数、每个消费者的待处理消息数。

2. 详细查看 待处理消息。使用 XPENDING 的范围查询格式。

   XPENDING mystream mygroup - + 10

• - +：表示ID范围从最小到最大。
• 10：最多返回10条记录。
  输出会列出每条消息的ID、所属消费者、空闲时间（自上次投递后的毫秒数）和投递次数。

第三步：确认（ACK）消息与故障恢复

这是实现可靠消费的核心步骤。成功处理消息后，必须发送确认。

1. 处理 并 确认 消息。假设 consumer1 成功处理了第一条消息。

   XACK mystream mygroup 1662543902000-0

• 用实际消息ID替换 1662543902000-0。
• XACK 命令会将该消息从 mygroup 的待处理列表中移除。之后，该组将无法再获取到这条消息。

2. 模拟 消费者故障。假设 consumer1 在读取消息后、确认前崩溃。

• 此时，该消息仍在 mygroup 的待处理列表中，且其空闲时间会不断增加。

3. 恢复 故障消息。其他消费者（如 consumer2）可以通过 认领 方式接管超时未确认的消息。

   XCLAIM mystream mygroup consumer2 3600000 1662543902000-0

• 3600000：空闲时间阈值（毫秒）。只有空闲时间超过1小时（示例值）的消息才会被认领。
• 1662543902000-0：指定要认领的消息ID。也可以使用 JUSTID 关键字来只返回ID。
• 执行后，该消息的所有权将转移到 consumer2，并重置其空闲时间。consumer2 现在可以正常处理并确认它。

4. 自动化 故障转移。更实用的方法是使用 XAUTOCLAIM，它自动认领一批超时消息。

   XAUTOCLAIM mystream mygroup consumer2 3600000 0-0 COUNT 10

• 0-0：从ID 0-0 开始扫描待处理列表。
• COUNT 10：尝试认领最多10条超时消息。
• 该命令会返回一个包含两部分的结果：成功认领的消息列表，以及下一个用于继续扫描的起始ID。

第四步：检查流与消费者组信息

使用 XINFO 系列命令进行监控。

1. 查看 流 mystream 的整体信息。

   XINFO STREAM mystream

2. 查看 mygroup 消费者的详细状态。

   XINFO CONSUMERS mystream mygroup

   输出会列出每个消费者（consumer1, consumer2）及其待处理消息数 pending 和空闲时间 idle。

应用场景与最佳实践

确保消息至少被处理一次：消费逻辑必须是幂等的。因为 XCLAIM 或服务重启可能导致消息被重复投递。设计 业务逻辑，使其能安全地处理重复消息。

处理消息堆积：当消费者速度慢于生产者时，待处理列表会增长。使用 XPENDING 和 XINFO 监控 消息堆积情况和消费者活跃度。考虑增加消费者实例，或优化单个消费者的处理速度。

设置合理的超时：XCLAIM 和 XAUTOCLAIM 中的空闲时间阈值，应大于消费者预期的最大处理时间。过小会导致消息在处理中被错误认领；过大会延长故障恢复时间。

结合使用 > 与 0：

• >：仅读取新消息，用于正常消费流程。
• 0：用于重新消费或检查组内所有未确认消息（包括待处理的）。例如：XREADGROUP GROUP mygroup consumer1 STREAMS mystream 0。这可以帮助消费者在重启后，恢复其未确认的消息列表。