Redis Stream 消费者组详解

什么是消费者组

Redis Stream 的消费者组（Consumer Group）是 Stream 实现消息队列的核心机制之一。它允许多个消费者协同消费同一个 Stream 中的消息，实现消息的负载均衡和容错处理。这个概念借鉴了 Kafka 的消费者组模型，但在 Redis 中有自己独特的实现方式。

消费者组的核心概念

1. 消费者组的组成

一个消费者组包含以下要素：

• 组名称：在 Stream 上唯一标识一个消费者组
• 游标（last_delivered_id）：组内已分发的最后一条消息 ID
• 待确认列表（Pending Entries List, PEL）：已投递但尚未确认的消息列表
• 消费者成员：属于该组的消费者，每个消费者有自己的 PEL

2. 创建消费者组

使用 XGROUP CREATE 命令创建消费者组：

XGROUP CREATE mystream mygroup $   # 从最新消息开始消费
XGROUP CREATE mystream mygroup 0   # 从 Stream 开头开始消费

$ 表示只消费组创建后的新消息，0 表示从 Stream 的第一条消息开始。

3. 消费者组的消息投递

消费者组的工作流程：

生产者 -> XADD -> Stream -> 消费者组 -> 分发给组内消费者

• 每条消息只会被投递给组内的 一个 消费者
• Redis 使用轮询（round-robin）算法进行消息分发
• 消费者通过 XREADGROUP 命令读取属于自己的消息

消费者组的优势

1. 负载均衡

# 消费者 A
XREADGROUP GROUP mygroup consumerA COUNT 1 STREAMS mystream >
# 消费者 B
XREADGROUP GROUP mygroup consumerB COUNT 1 STREAMS mystream >

当多个消费者通过同一个消费者组读取时，Redis 会自动将消息分发给不同的消费者，实现负载均衡。

2. 消息追踪与容错

消费者组维护了每个消费者的 PEL（Pending Entries List），记录了已投递但未确认的消息。这为消息的可靠消费提供了基础：

• 消费者崩溃后重启，可以从 PEL 中获取未处理的消息
• 使用 XPENDING 命令查看待处理消息
• 使用 XACK 命令确认消息处理完成

3. 消息回溯

消费者组支持从任意位置重新消费：

# 从指定 ID 开始重新消费
XREADGROUP GROUP mygroup consumerB STREAMS mystream 0

这在故障恢复和消息重试场景中非常有用。

消费者组的关键命令

消费者组的使用模式

基本消费模式

# Python 伪代码
while True:
    messages = redis.xreadgroup('mygroup', 'consumer1', {'mystream': '>'}, count=10, block=2000)
    for msg in messages:
        process(msg)
        redis.xack('mystream', 'mygroup', msg['id'])

故障恢复模式

# 读取待处理的消息
pending = redis.xpending('mystream', 'mygroup')
if pending['pending'] > 0:
    # 获取待处理消息详情
    msgs = redis.xpending_range('mystream', 'mygroup', '-', '+', 10)
    for msg in msgs:
        if msg['times_delivered'] < 3:
            # 重新处理
            process(msg)
            redis.xack('mystream', 'mygroup', msg['message_id'])
        else:
            # 超过重试次数，转移到死信队列
            dead_letter_queue(msg)

消费者组的限制

1. 消息持久性：消费者的偏移量存储在内存中，不持久化到 RDB/AOF
2. 消费者组数量：每个 Stream 可以有多个消费者组，但过多会影响性能
3. 消息积压：消费者处理速度跟不上生产者时，PEL 会持续膨胀，占用内存

面试要点

• 消费者组 vs 普通读取：XREAD 是独立消费，XREADGROUP 是组内协作消费
• 游标位置：> 表示读取未分发的消息，数字 ID 表示从指定位置开始读取（包括已分发的）
• 消息ACK：XACK 是核心——消费者组只保证 at-least-once，需要 ACK 来确认处理完成
• PEL 膨胀问题：长时间不 ACK 会导致 PEL 占用大量内存，需要设置监控预警