Redis xreadgroup命令解析及Python消费者代码疑问解答

逐行代码说明 #

1. 创建消费者组：redis.xgroup_create(key, group) #

这行代码的作用就是为order_completed这个Redis Stream创建名为inventory_group的消费者组。消费者组的核心价值是实现多消费者协作消费同一条Stream的消息，保证每条消息只会被组内的一个消费者处理，避免重复消费。

2. 组内读取消息：redis.xreadgroup(group, key, {key: '>'}, None) #

这行是通过消费者组读取Stream消息的核心逻辑，重点解释你疑惑的{key: '>'}：

• 这个字典的键是目标Stream的名称（即order_completed），值'>'是Redis Stream的特殊ID标记，含义是：让当前消费者读取该消费者组从未处理过的最新消息。
• 消费者组会维护一个「已投递消息ID指针」，记录组内所有消费者已经处理到哪个消息ID。使用'>'时，Redis会跳过所有已经被组内其他消费者处理过的消息，直接返回指针之后的新消息（包括调用后刚产生的消息）。
• 举个对比例子：如果这里填具体的消息ID（比如1620000000000-0），会从该ID之后开始读取；如果填'0'，则会读取Stream里的所有历史消息（无论是否被组内处理过）。

额外补充代码里的其他参数：

• 第二个参数key在这里是消费者的名称（你这里复用了Stream键的名字，其实可以自定义，比如'inventory_consumer_01'）；
• 最后一个参数None表示无超时阻塞，会一直等待直到有新消息产生才返回。

内容的提问来源于stack exchange，提问作者meallhour