【Redis】发布订阅与消息队列Day8（2026年）

写在前面

消息队列是分布式系统中不可或缺的组件，Redis除了作为缓存数据库，还提供了发布订阅（Pub/Sub）和Stream两种消息机制。虽然Redis不是专业的消息队列，但在轻量级场景下，它是一个不错的选择。今天我们来深入理解Redis的消息能力。

文章目录

• • 写在前面
  • 一、Redis发布订阅（Pub/Sub）
  • • 1.1 什么是发布订阅？
    • 1.2 Pub/Sub基本命令
    • 1.3 Pub/Sub使用示例
    • 1.4 模式订阅
  • 二、发布订阅的局限性
  • • 2.1 主要局限性
    • 2.2 适用场景
  • 三、Redis Stream数据结构
  • • 3.1 什么是Stream？
    • 3.2 Stream基本命令
    • 3.3 Stream使用示例
  • 四、Stream消费组
  • • 4.1 消费组概念
    • 4.2 消费组操作
    • 4.3 消费组配置
  • 五、Stream vs Kafka对比
  • 六、踩坑提醒
  • 七、实际应用案例
  • • 7.1 实时消息推送
    • 7.2 日志收集
    • 7.3 事件溯源
  • 八、面试高频考点
  • • 考点1：Redis做消息队列的优缺点？
    • 考点2：Pub/Sub和Stream的区别？
    • 考点3：如何保证消息不丢失？
    • 考点4：Stream的消息ID有什么特点？
  • 九、参考资料
  • 十、互动话题

一、Redis发布订阅（Pub/Sub）

1.1 什么是发布订阅？

实际场景：即时通讯系统中，用户上线后需要订阅多个群组消息，当有人在群组发言时，所有在线成员都能实时收到消息。

发布订阅（Pub/Sub）是一种消息通信模式：

• 发布者（Publisher）：发送消息到频道
• 订阅者（Subscriber）：订阅频道接收消息
• 频道（Channel）：消息传递的通道

┌────────────┐
│ Publisher │──────┐
└────────────┘ │
▼
┌──────────────┐
│ Channel │
└──────────────┘
│
┌───────────┼───────────┐
▼ ▼ ▼
┌────────────┐ ┌────────────┐ ┌────────────┐
│ Subscriber │ │ Subscriber │ │ Subscriber │
└────────────┘ └────────────┘ └────────────┘

1.2 Pub/Sub基本命令

1.3 Pub/Sub使用示例

订阅消息：

# 客户端1：订阅频道
SUBSCRIBE news:sports
# 返回：
# 1) "subscribe"
# 2) "news:sports"
# 3) (integer) 1

# 进入阻塞状态，等待消息

发布消息：

# 客户端2：发布消息
PUBLISH news:sports "NBA总决赛：湖人vs凯尔特人"
# 返回：(integer) 1 表示有1个订阅者收到消息

订阅者收到消息：

# 客户端1收到：
# 1) "message"
# 2) "news:sports"
# 3) "NBA总决赛：湖人vs凯尔特人"

1.4 模式订阅

使用通配符订阅多个频道：

# 订阅所有news开头的频道
PSUBSCRIBE news:*

# 可以匹配：
# news:sports, news:tech, news:weather 等

二、发布订阅的局限性

踩坑提醒：Redis Pub/Sub是实时的，如果订阅者离线，消息会丢失！

2.1 主要局限性

2.2 适用场景

三、Redis Stream数据结构

3.1 什么是Stream？

经验之谈：Redis 5.0引入的Stream是专门为消息队列设计的数据结构，解决了Pub/Sub的主要痛点。

Stream是Redis 5.0引入的数据结构，具有以下特点：

• 消息持久化存储
• 支持消费组
• 支持消息确认（ACK）
• 支持消息回溯
• 类似Kafka的设计理念

3.2 Stream基本命令

3.3 Stream使用示例

添加消息：

# 添加消息到Stream
XADD mystream * name "zhangsan" age 25
# 返回：1640000000000-0（消息ID）

# * 表示由Redis自动生成ID
# 格式：毫秒时间戳-序列号

读取消息：

# 从头开始读取
XRANGE mystream – +
# 返回所有消息

# 读取最新消息（非阻塞）
XREAD COUNT 2 STREAMS mystream 0

# 阻塞读取最新消息
XREAD BLOCK 5000 STREAMS mystream $
# $表示最新消息位置，BLOCK 5000表示阻塞5秒

范围查询：

# 按时间范围查询
XRANGE mystream 1640000000000 1640000001000

# 只取2条
XRANGE mystream – + COUNT 2

四、Stream消费组

4.1 消费组概念

消费组（Consumer Group）是Stream的核心特性：

• 多个消费者组成一个组
• 每条消息只能被组内一个消费者消费
• 支持消息确认和重新消费

┌──────────────┐
│ Stream │
└──────────────┘
│
┌────────────┼────────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│Consumer-1│ │Consumer-2│ │Consumer-3│
└──────────┘ └──────────┘ └──────────┘
│ │ │
└────────────┼────────────┘
│
┌──────────────┐
│ Consumer Group│
└──────────────┘

4.2 消费组操作

创建消费组：

# 从头开始消费
XGROUP CREATE mystream mygroup 0

# 只消费新消息
XGROUP CREATE mystream mygroup $

# 创建成功后查看信息
XINFO GROUPS mystream

消费组读取消息：

# 消费组读取
XREADGROUP GROUP mygroup consumer1 COUNT 1 STREAMS mystream >

# > 表示读取未消费的消息
# consumer1是消费者名称

确认消息：

# 处理完消息后确认
XACK mystream mygroup 1640000000000-0
# 返回：(integer) 1

查看待处理消息：

# 查看待处理消息
XPENDING mystream mygroup

# 查看详细信息
XPENDING mystream mygroup – + 10

转移消息所有权：

# 将消息转移给其他消费者
XCLAIM mystream mygroup consumer2 0 1640000000000-0

4.3 消费组配置

# 设置最大长度，防止Stream无限增长
XADD mystream MAXLEN 1000 * field value

# 精确限制长度（性能更好）
XADD mystream MAXLEN ~ 1000 * field value

五、Stream vs Kafka对比

六、踩坑提醒

踩坑提醒：消息丢失问题

问题1：Pub/Sub消息丢失

# 订阅者离线时，消息直接丢失
SUBSCRIBE channel
# 离线期间发布的消息无法收到

解决方案：

• 使用Stream替代Pub/Sub
• 使用专业消息队列（RabbitMQ、Kafka）

问题2：Stream消息堆积

# Stream无限增长会耗尽内存
XADD stream * data value
# 持续添加不删除

解决方案：

# 使用MAXLEN限制长度
XADD stream MAXLEN ~ 10000 * data value

# 定期清理
XTRIM stream MAXLEN 10000

问题3：消费者宕机消息未确认

# 消费者读取消息后宕机，消息处于pending状态
XREADGROUP GROUP mygroup consumer1 STREAMS stream >
# 宕机…

解决方案：

# 其他消费者接管pending消息
XCLAIM stream mygroup consumer2 60000 pending-message-id
# 60000是消息空闲时间（毫秒）

七、实际应用案例

7.1 实时消息推送

# 发布者
XADD notifications * user_id 1001 content "您有新消息"

# 消费者
XREADGROUP GROUP notify_group consumer1 BLOCK 5000 STREAMS notifications >

7.2 日志收集

# 应用服务写入日志
XADD app:logs * level ERROR service user-api message "连接超时"

# 日志处理服务消费
XREADGROUP GROUP log_group log_processor STREAMS app:logs >

7.3 事件溯源

# 记录领域事件
XADD order:events * event OrderCreated order_id 1001 amount 99.9
XADD order:events * event OrderPaid order_id 1001 paid_at 1640000000
XADD order:events * event OrderShipped order_id 1001 shipped_at 1640000100

# 回放事件
XRANGE order:events – +

八、面试高频考点

考点1：Redis做消息队列的优缺点？

答案：

考点2：Pub/Sub和Stream的区别？

答案：

考点3：如何保证消息不丢失？

答案：

考点4：Stream的消息ID有什么特点？

答案：

• 格式：毫秒时间戳-序列号，如1640000000000-0
• 时间戳由Redis服务器生成，保证递增
• 同一毫秒内通过序列号区分
• 支持自定义ID，但通常使用*让Redis自动生成

九、参考资料

十、互动话题

欢迎在评论区分享你的经验和见解！

下一期预告：Day9 – Redis主从复制，敬请期待！