Redis 消息队列学习笔记
一、Redis 消息队列的三种实现方式
Redis 可以通过三种方式实现消息队列：
List：基于列表实现简单消息队列。
Pub/Sub：基于发布订阅实现消息广播。
Stream：支持消费者组、消息确认和故障恢复。
1. List 消息队列
List 可以通过 RPUSH 和 LPOP 实现先进先出的消息队列。
RPUSH queue msg1 # 生产者发送消息
LPOP queue # 消费者获取消息
BLPOP queue 0 # 队列为空时阻塞等待特点：
支持多个生产者和多个消费者。
支持阻塞读取。
不具备原生 ACK 确认机制。
消息弹出后就从 List 中移除，消费者宕机可能导致消息丢失。
注意： List 并不是只能单生产者、单消费者。多个消费者可以竞争消费同一个 List，但每条消息通常只会被其中一个消费者弹出。
2. Pub/Sub（发布订阅）
生产者通过 PUBLISH 发布消息，消费者通过 SUBSCRIBE 订阅频道。
SUBSCRIBE news # 订阅频道
PUBLISH news "hello" # 发布消息特点：
支持多个生产者和多个消费者。
一条消息可以广播给多个在线订阅者。
不持久化消息，也没有 ACK 确认机制。
消费者不在线时，无法收到之前发送的消息。
消费者处理速度过慢，可能因输出缓冲区达到限制而断开连接，导致消息丢失。
适用场景： 实时通知、消息广播等对消息可靠性要求不高的业务。
3. Stream 消息队列
Stream 是 Redis 5.0 引入的数据结构，提供了更完善的消息队列功能。
# 生产消息
XADD mq:orders * orderId 1001
# 读取消息
XREAD COUNT 1 BLOCK 2000 STREAMS mq:orders 0特点：
支持消息持久化（取决于 Redis 持久化配置）。
支持阻塞读取。
支持消费者组。
支持 ACK 消息确认。
支持 Pending List 记录未确认消息。
支持历史消息回溯。
注意：Stream 并不保证绝对不丢消息，仍需要考虑 Redis 持久化、消息裁剪和消费者故障等问题。
二、三种消息队列对比
总结：
List：适合简单队列。
Pub/Sub：适合实时广播。
Stream：适合需要消息确认和故障恢复的队列。
三、Stream 消费者组
消费者组（Consumer Group）允许多个消费者共同处理同一个 Stream 中的消息。
主要有三个核心特点。
1. 消息分流
同一个消费者组内，多个消费者共同消费消息。
Stream
|
Consumer Group
|
┌────────┼────────┐
↓ ↓ ↓
消费者A 消费者B 消费者C同一条新消息通常分配给组内一个消费者处理，避免重复分发给所有消费者。
作用：提高消息处理能力，降低消息堆积风险。
不同消费者组之间有各自的消费进度，可以分别消费同一条消息。
2. 维护消费进度
Redis 会为消费者组记录最后投递的消息 ID。
当消费者读取新消息时，会根据消费组进度继续投递。
注意：记录的是消息投递进度，不是最后成功处理的消息。
如果消费者宕机，已经投递但未确认的消息不会自动作为新消息重新投递，需要通过 Pending List 恢复。
3. ACK 消息确认
消费者组读取消息后，消息会进入 Pending List。
生产者 XADD
↓
Stream 保存消息
↓
消费者 XREADGROUP
↓
消息进入 Pending List
↓
消费者处理业务
↓
处理成功 XACK
↓
从 Pending List 移除重要：XACK 只会将消息从 Pending List 中移除，不会删除 Stream 中保存的原始消息。
四、Pending List 与消息恢复
Pending List（PEL）用于记录已经投递给消费者、但尚未 ACK 的消息。
1. 为什么需要 Pending List？
假设消费者 A 获取消息后突然宕机：
消费者 A 获取消息
↓
消息进入 Pending List
↓
消费者 A 宕机
↓
消息没有 XACK
↓
仍然保留 Pending 状态如果没有恢复机制，这些消息可能长期得不到处理。
2. 常用命令
创建消费者组
XGROUP CREATE mq:orders g1 0 MKSTREAM0 表示从 Stream 中最早的现存消息开始消费。
消费者读取新消息
XREADGROUP GROUP g1 c1 COUNT 10 BLOCK 2000 STREAMS mq:orders >其中：
g1：消费者组。c1：消费者名称。>：读取尚未投递给该组的新消息。
查看未确认消息
XPENDING mq:orders g1重新认领超时未确认的消息
XAUTOCLAIM mq:orders g1 c2 60000 0-0 COUNT 10表示由消费者 c2 认领空闲至少 60 秒的 Pending 消息。
确认消费完成
XACK mq:orders g1 消息ID业务执行成功后，使用 XACK 移除 Pending 记录。
五、Stream 如何处理消费者宕机？
完整流程：
消费者 A 获取消息
↓
消息进入 Pending List
↓
消费者 A 宕机
↓
未执行 XACK
↓
其他消费者通过 XPENDING 检查
↓
使用 XAUTOCLAIM 认领超时消息
↓
重新处理业务
↓
处理成功后 XACK这样可以避免消费者宕机后消息一直无人处理。
注意：Redis 不会自动保证每条 Pending 消息都得到处理，需要应用程序实现恢复和重试逻辑。
六、Stream 的两个注意事项
1. 消息可能重复消费
例如：
消费者 A 已经完成数据库更新。
但还没执行
XACK就宕机。消费者 B 重新认领该消息。
相同业务可能再次执行。
因此，消费者必须考虑幂等性。
例如订单业务，可以根据订单 ID 防止重复创建订单。
2. 消息可能漏处理
使用 XREADGROUP ... > 只能读取新消息，不会自动读取其他消费者遗留的 Pending 消息。
因此，需要配合：
XPENDING：检查未确认消息。XAUTOCLAIM：重新认领超时消息。XACK：消费成功后确认。
同时，应避免错误设置消费起点或过早裁剪未处理消息。
七、总结
Redis 消息队列有三种实现方式：List、Pub/Sub 和 Stream。
List 可以通过 RPUSH、LPOP、BLPOP 实现简单队列，但没有原生消息确认机制。
Pub/Sub 通过 PUBLISH 和 SUBSCRIBE 实现消息广播，但不支持消息持久化和 ACK，消费者离线时容易丢失消息。
Stream 功能更加完善，支持消费者组、消息确认和历史消息读取。消费者组可以实现组内消息分流，并维护消费进度。
消息投递后进入 Pending List，处理成功通过 XACK 确认。如果消费者宕机，可以通过 XPENDING 检查、XAUTOCLAIM 认领未确认消息，实现故障恢复。
但 Stream 仍可能发生重复消费，因此业务需要做好幂等处理。
核心记忆：List 实现简单队列，Pub/Sub 实现广播，Stream 通过消费者组、Pending List 和 ACK 实现更可靠的消息消费。