如何利用Redis Stream实现按Key路由消息至单消费者组？

方案原理与需求匹配 #

你的需求可以通过Redis Stream消费者组配合Redis分布式锁的组合来实现，以下是对应每个需求的具体实现逻辑：

1. 满足「单消费者拾取消息，不广播」+「同一ID消息不重复处理」

直接使用Redis Stream的消费者组模式：

• 创建消费者组：XGROUP CREATE <stream-name> <group-name> $ MKSTREAM（$表示从流的末尾开始消费，MKSTREAM自动创建流）
• 消费者通过XREADGROUP GROUP <group-name> <consumer-id> COUNT 1 BLOCK 5000 STREAMS <stream-name> >获取消息（>表示消费未被组内任何消费者认领的消息）

消费者组的原生特性确保：

• 每条消息只会被分配给组内的一个消费者，不会广播给所有消费者
• 未被XACK确认的消息会留在「待处理队列」中，不会被其他消费者重复认领，直到被确认或通过XCLAIM转移

2. 满足「不同Key消息并行处理」+「同一Key消息独占处理」

在消费者获取消息后，新增基于消息Key的分布式锁逻辑，流程如下：

1. 消费者通过XREADGROUP拿到消息后，提取消息中的Key字段（比如K1）
2. 尝试获取该Key的独占锁：

   SET lock:K1 <consumer-id> NX PX 30000
   

   • NX表示仅当锁不存在时才创建
   • PX 30000设置锁的过期时间为30秒（防止消费者崩溃导致锁永久占用，时间可根据业务处理时长调整）
3. 锁获取结果分支：
   • 成功获取锁：正常处理消息，处理完成后先执行XACK <stream-name> <group-name> <message-id>确认消息，再删除锁：DEL lock:K1
   • 锁获取失败：直接放弃这条消息（不处理也不ACK），让它留在待处理队列中，等待持有该Key锁的消费者后续处理（或锁过期后重新竞争）

这种机制保证：

• 不同Key的锁可以被不同消费者同时持有，实现并行处理
• 同一Key的消息只能被持有对应锁的消费者处理，避免数据覆盖

3. 补充细节优化

• 锁的安全性：如果担心锁过期但消费者还在处理消息，可以实现「锁续约」逻辑（比如每隔10秒重新设置锁的过期时间）
• 超时消息处理：对于长时间未被ACK的消息，可定期用XCLAIM命令将其转移给其他消费者，避免消息堆积：

  XCLAIM <stream-name> <group-name> <consumer-id> 60000 COUNT 10 STREAMS <stream-name> 0
  

  （将60秒未被处理的消息转移给当前消费者）
• 消费者崩溃后的处理：锁过期后，其他消费者可以竞争获取锁并处理同Key的消息，同时XCLAIM可以认领崩溃消费者未处理的消息，保证消息不丢失

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