Redis Pub/Sub 消息丢失的原因与 Stream 的持久化替代

Redis 的 Pub/Sub（发布/订阅）模式是构建实时消息系统的常用工具。它允许发布者将消息发送到频道，而所有订阅该频道的订阅者都能收到消息。然而，它有一个致命的弱点：消息可能丢失。本文将深入分析 Redis Pub/Sub 消息丢失的根本原因，并介绍 Redis Stream 作为一种更可靠的持久化替代方案。

Redis Pub/Sub 的工作原理

在了解问题之前，我们先快速回顾一下 Redis Pub/Sub 的工作方式。它是一个典型的发布-订阅模型，发布者将消息发送到一个或多个频道，订阅者则订阅这些频道以接收消息。

这种模式的特点是实时性和解耦。发布者无需知道订阅者的存在，订阅者也可以在不影响发布者的情况下动态加入或离开。但正是这种“无状态”的特性，导致了消息丢失的问题。

Redis Pub/Sub 消息丢失的根本原因

Redis Pub/Sub 模式本身不保证消息的持久化。消息在发布后，只会存在于 Redis 服务器的内存中，并立即转发给所有当前在线的订阅者。一旦出现以下情况，消息就会丢失：

1. 订阅者离线时
   如果一个订阅者在消息发布时处于离线状态（例如，网络中断或进程崩溃），它将无法收到该消息。当它重新连接后，Redis 不会重发之前错过的消息。

2. 网络抖动或分区
   即使订阅者短暂断开连接后又迅速恢复，在这期间发布的消息也会丢失。Redis Pub/Sub 没有机制来存储这些“中间”消息并稍后重发。

3. Redis 服务器重启
   Redis 的默认持久化策略（RDB 或 AOF）不会保存 Pub/Sub 的消息。如果 Redis 服务器重启，所有内存中的 Pub/Sub 状态（包括待处理的消息）都会被清空，导致消息丢失。

4. 订阅者进程崩溃
   如果一个订阅者进程在处理消息的过程中崩溃，它没有机会确认消息，这条消息也就丢失了。Pub/Sub 模式没有“已读确认”机制。

解决方案：Redis Stream 的持久化替代

为了解决上述问题，Redis 提供了 Stream 数据结构。Stream 是一个持久化的、有结构的消息队列，它从根本上解决了 Pub/Sub 的消息丢失问题。Stream 的核心思想是为每个消息分配一个唯一 ID，并允许消费者以组的形式消费消息，同时需要显式地确认（ACK）消息。

Stream 的关键概念

• 消息 ID (Message ID)：每条消息都有一个唯一的 ID，格式为 timestamp-sequence number（例如 1640995200000-0）。这确保了消息的有序性。
• 消费者组 (Consumer Group)：一组消费者共同消费一个 Stream。每个组都有一个名称和偏移量，记录了该组已处理的消息位置。
• 消费者 (Consumer)：消费者组中的单个成员。每个消费者有自己的名称和偏移量，记录了自己已处理的消息位置。

使用 Stream 实现可靠消息传递的实践指南

以下是使用 Redis Stream 作为 Pub/Sub 替代方案的步骤。

1. 创建一个 Stream

首先，我们需要创建一个 Stream。可以使用 XADD 命令向 Stream 中添加消息。

# 创建一个名为 'my_stream' 的 Stream，并添加一条消息
XADD my_stream * name "Alice" age 30

• my_stream：Stream 的名称。
• *：表示让 Redis 自动生成消息 ID。你也可以指定一个自定义的 ID。

2. 创建一个消费者组

为了实现组消费和消息确认，我们需要为 Stream 创建一个消费者组。

# 为 'my_stream' 创建一个名为 'my_group' 的消费者组，'MKSTREAM' 选项表示如果 Stream 不存在则自动创建
XGROUP CREATE my_stream my_group $ MKSTREAM
```

*   `my_group`：消费者组的名称。
*   `$`：表示从 Stream 的末尾开始消费，即只消费新消息。
*   `MKSTREAM`：如果 Stream 不存在，则创建它。

#### 3. 作为生产者发布消息

生产者使用 `XADD` 命令向 Stream 持续发布消息，这与 Pub/Sub 类似，但消息会被持久化。

```bash
# 向 'my_stream' 添加一条新消息
XADD my_stream * message "Hello Stream"

4. 作为消费者消费消息

消费者使用 XREADGROUP 命令从其所属的消费者组中读取消息。

# 以 'consumer1' 的身份，从 'my_group' 消费组中读取 1 条消息
XREADGROUP GROUP my_group consumer1 COUNT 1 STREAMS my_stream >

• GROUP my_group：指定消费的消费者组。
• consumer1：指定消费者的名称。
• COUNT 1：每次读取的消息数量。
• STREAMS my_stream >：指定要读取的 Stream，> 表示从该消费者上次读取的位置继续（即只读取新消息）。

5. 确认消息 (ACK)

这是 Stream 与 Pub/Sub 最关键的区别。消费者在成功处理消息后，必须使用 XACK 命令向 Stream 确认该消息。只有被确认的消息，才会从 Stream 中被标记为已处理，并可以被后续的消费者读取。

# 假设从上一步读取到的消息 ID 是 '1640995200000-1'
XACK my_stream my_group 1640995200000-1

• my_stream：Stream 名称。
• my_group：消费者组名称。
• 1640995200000-1：要确认的消息 ID。

通过这种方式，即使消费者在处理消息时崩溃，Redis 也会在消费者重新连接后，将未确认的消息重新分配给其他消费者，从而确保消息不会丢失。

Pub/Sub vs. Stream 特性对比

为了更清晰地理解两者的差异，我们可以通过一个表格进行对比。

通过以上分析，我们可以看到 Redis Stream 提供了比 Pub/Sub 更强大的功能和可靠性，特别是在需要保证消息不丢失的关键业务场景中。虽然 Stream 的使用比 Pub/Sub 稍微复杂一些，但其带来的持久化和可靠性优势使其成为现代应用中更值得推荐的选择。