Redis - 发布订阅

Redis - 发布/订阅（Pub/Sub）和流

Redis 提供了强大的消息传递范式。最初的“即发即忘”（fire-and-forget）模型是 Pub/Sub，而较新、更强大的模型是 Redis Streams，它提供了一个持久的、只追加的日志。

经典 Pub/Sub

Redis Pub/Sub 是一种消息系统，其中发送者（发布者 Publisher）将消息发送到命名“通道”（channels），而无需知道有哪些订阅者（Subscriber）。接收者（订阅者）表示对一个或多个通道感兴趣，并且仅在其连接并订阅时接收消息。

Pub/Sub 的主要特性

• 即发即忘：如果在发布消息时没有订阅者监听某个通道，则消息将永久丢失。
• 无历史记录：订阅者看不到在其订阅之前发布的消息。
• 解耦：发布者和订阅者完全解耦，彼此之间互不感知。

redis-cli 示例

想象两个独立的客户端会话连接到同一个 Redis 服务器。

-- Client A (Subscriber) --

redis> SUBSCRIBE live-updates
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "live-updates"
3) (integer) 1

-- At this point, Client A is blocked, waiting for messages --

-- Client B (Publisher) --

redis> PUBLISH live-updates "System rebooting in 5 minutes!"
(integer) 1

redis> PUBLISH live-updates "Reboot complete. All services are back online."
(integer) 1

-- Client A's console now shows the received messages --

1) "message"
2) "live-updates"
3) "System rebooting in 5 minutes!"

1) "message"
2) "live-updates"
3) "Reboot complete. All services are back online."

更强大的替代方案：Redis Streams

Streams 于 Redis 5.0 引入，是一种持久的、只追加的日志数据结构。它们提供了一个更强大的消息系统，适用于消息队列、事件溯源和持久通知等用例。

Streams 的主要特性

• 持久性：消息存储在 Redis 服务器的流中，直到明确删除。消费者可以从历史记录的任何点读取消息。
• 消费者组（Consumer Groups）：允许多个客户端协同工作以消费单个流中的消息。Redis 确保组内的每条消息只传递给一个消费者。
• 消息确认（Message Acknowledgment）：消费者必须明确确认（XACK）他们已经处理了一条消息。这允许“至少一次”交付和从消费者故障中恢复。
• 阻塞式读取：类似于列表的 BRPOP，客户端可以阻塞直到流中有新消息可用。

Stream 示例（redis-cli）

-- Producer: Add a message to the 'orders' stream --
# 生产者：向 'orders' 流添加一条消息
# The '*' tells Redis to auto-generate a unique ID (timestamp-sequence)
# '*' 告诉 Redis 自动生成一个唯一的 ID（时间戳-序列）

redis> XADD orders * user_id 123 product_id 987 quantity 2
"1672531200000-0"

-- Consumer: Read all messages from the beginning of the stream --
# 消费者：从流的开头读取所有消息
# The '0-0' is a special ID meaning 'the very first entry'
# '0-0' 是一个特殊 ID，表示“第一个条目”

redis> XREAD COUNT 100 STREAMS orders 0-0
1) 1) "orders"
   2) 1) 1) "1672531200000-0"
         2) 1) "user_id"
            2) "123"
            3) "product_id"
            4) "987"
            5) "quantity"
            6) "2"

Pub/Sub 与 Streams：如何选择？

• 使用 Pub/Sub 适用于：瞬时、实时通知，可接受消息丢失。示例：实时聊天室通知，通知系统缓存失效。
• 使用 Streams 适用于：持久、可靠的消息传递，其中每条消息都必须被处理。示例：处理订单提交、作业队列、微服务事件日志。

相关命令

Pub/Sub 命令

命令	描述
SUBSCRIBE channel [channel …]	监听发布到给定通道的消息。
PUBLISH channel message	向通道发布消息。
PSUBSCRIBE pattern [pattern …]	订阅与 glob 风格模式匹配的通道。
UNSUBSCRIBE [channel …]	停止监听给定通道上的消息。

核心 Stream 命令

命令	描述
XADD key ID field string [field string …]	向流中追加一个新的条目。
XREAD [COUNT count] [BLOCK ms] STREAMS key [key …] ID [ID …]	从一个或多个流中读取一个或多个条目，可选地阻塞。
XGROUP CREATE key groupname ID [MKSTREAM]	创建一个与流关联的新消费者组。
XREADGROUP GROUP group consumer [COUNT count] [BLOCK ms] STREAMS key [key …] ID [ID …]	作为消费者组的一部分从流中读取。
XACK key group ID [ID …]	确认一个或多个消息已成功处理。