3.2 发布订阅 (PubSub)

文档摘要

3.2 发布订阅 (Pub/Sub) Redis 高级特性：发布订阅 (Pub/Sub) 详解与实践 Redis 除了作为高性能的键值数据库外，还提供了许多强大的高级特性。其中，发布订阅 (Publish/Subscribe) 机制是一种强大的消息通信模式，它允许消息的发送者（发布者）将消息发送到频道，而无需知道哪些接收者（订阅者）正在监听这些频道。订阅者可以订阅一个或多个频道，接收所有发布到这些频道的消息。发布订阅模式在构建实时性要求高的应用中非常有用，例如：实时消息推送: 例如聊天室、实时评论、新闻推送等。事件通知: 系统事件的广播，例如缓存失效通知、配置更新等。微服务架构: 作为服务间异步通信的一种方式。日志系统: 收集和分发日志信息。

3.2 发布订阅 (Pub/Sub)

Redis 高级特性：发布订阅 (Pub/Sub) 详解与实践

Redis 除了作为高性能的键值数据库外，还提供了许多强大的高级特性。其中，发布订阅 (Publish/Subscribe) 机制是一种强大的消息通信模式，它允许消息的发送者（发布者）将消息发送到频道，而无需知道哪些接收者（订阅者）正在监听这些频道。订阅者可以订阅一个或多个频道，接收所有发布到这些频道的消息。

发布订阅模式在构建实时性要求高的应用中非常有用，例如：

实时消息推送: 例如聊天室、实时评论、新闻推送等。
事件通知: 系统事件的广播，例如缓存失效通知、配置更新等。
微服务架构: 作为服务间异步通信的一种方式。
日志系统: 收集和分发日志信息。

1. 发布订阅模式的核心概念

在 Redis 的 Pub/Sub 模式中，主要有以下几个核心概念：

发布者 (Publisher): 消息的发送者。发布者将消息发送到指定的频道，无需关心有多少订阅者以及订阅者是谁。
订阅者 (Subscriber): 消息的接收者。订阅者可以订阅一个或多个频道，接收发布到这些频道的所有消息。
频道 (Channel): 消息的传输通道。发布者将消息发布到频道，订阅者从频道接收消息。频道名称是字符串，可以根据业务逻辑自定义命名。
消息 (Message): 发布者发送到频道的数据。消息可以是任何字符串或二进制数据。

工作流程简述:

发布者将消息 M 发布到频道 C。
Redis 服务器接收到消息 M 后，将消息 M 推送给所有订阅了频道 C 的订阅者。
订阅者接收到来自频道 C 的消息 M。

核心特点:

解耦: 发布者和订阅者之间完全解耦，发布者无需知道订阅者的存在。
实时性: 消息一旦发布，会立即推送给所有订阅者，延迟很低。
扇出 (Fan-out): 一个发布者可以向多个订阅者广播消息。
无持久化: Redis Pub/Sub 是一种“fire-and-forget”的机制，消息不会被持久化。如果订阅者在消息发布时离线，则会丢失该消息。
模式匹配订阅: 除了订阅特定频道外，还支持基于模式匹配的订阅，例如订阅 news.* 可以接收 news.sports、news.politics 等频道的消息。

2. Redis Pub/Sub 相关命令详解

Redis 提供了以下命令来支持发布订阅功能：

PUBLISH channel message: 将消息 message 发布到指定的 channel。
- 返回值: 接收到消息的订阅者数量。如果没有任何订阅者，则返回 0。
- 示例:
```
PUBLISH news.sports "Breaking news: Team X wins championship!"
```
SUBSCRIBE channel [channel ...]: 订阅一个或多个频道。
- 客户端进入订阅模式: 执行 SUBSCRIBE 命令后，客户端连接会进入订阅模式。在订阅模式下，客户端只能接收订阅相关的命令（如 SUBSCRIBE, PSUBSCRIBE, UNSUBSCRIBE, PUNSUBSCRIBE, PING, QUIT）。
- 接收消息: 订阅成功后，服务器会将发布到订阅频道的消息推送给客户端。
- 返回值: 服务器会返回订阅成功的确认信息，以及后续接收到的消息。消息格式是一个三元素的数组，第一个元素是消息类型（message），第二个元素是频道名称，第三个元素是消息内容。
- 示例:
```
SUBSCRIBE news.sports news.politics
```
  订阅 news.sports 和 news.politics 两个频道。
PSUBSCRIBE pattern [pattern ...]: 订阅与给定模式匹配的所有频道。
- 模式匹配: 模式支持通配符，例如 news.* 可以匹配 news.sports, news.politics, news.technology 等频道。
- 返回值: 与 SUBSCRIBE 类似，服务器会返回订阅成功的确认信息，以及后续接收到的消息。消息格式是一个四元素的数组，第一个元素是消息类型（pmessage），第二个元素是模式，第三个元素是频道名称，第四个元素是消息内容。
- 示例:
```
PSUBSCRIBE news.*
```
  订阅所有以 news. 开头的频道。
UNSUBSCRIBE [channel [channel ...]]: 取消订阅一个或多个频道。
- 不指定频道: 如果不指定频道，则取消订阅所有已订阅的频道。
- 客户端退出订阅模式: 如果取消订阅了所有频道，客户端会退出订阅模式，可以执行其他 Redis 命令。
- 返回值: 服务器会返回取消订阅成功的确认信息。
- 示例:
```
UNSUBSCRIBE news.sports
UNSUBSCRIBE  # 取消订阅所有频道
```
PUNSUBSCRIBE [pattern [pattern ...]]: 取消订阅与给定模式匹配的所有频道。
- 不指定模式: 如果不指定模式，则取消订阅所有已订阅的模式。
- 返回值: 服务器会返回取消订阅成功的确认信息。
- 示例:
```
PUNSUBSCRIBE news.*
PUNSUBSCRIBE  # 取消订阅所有模式
```
PUBSUB subcommand [argument [argument ...]]: 用于内省和管理 Pub/Sub 子系统。
- PUBSUB CHANNELS [pattern]: 列出活跃的频道。
  - pattern: 可选的模式，用于过滤频道名称。
  - 返回值: 活跃频道名称的列表。
  - 示例:
```
PUBSUB CHANNELS
PUBSUB CHANNELS news.*
```
- PUBSUB NUMPAT: 返回当前活跃的模式订阅数量。
  - 返回值: 整数，表示活跃的模式订阅数量。
  - 示例:
```
PUBSUB NUMPAT
```
- PUBSUB NUMSUB [channel [channel ...]]: 返回给定频道的订阅者数量。
  - 不指定频道: 如果不指定频道，则返回所有频道的订阅者数量之和。
  - 返回值: 整数，表示订阅者数量。
  - 示例:
```
PUBSUB NUMSUB news.sports news.politics
PUBSUB NUMSUB  # 返回所有频道的订阅者总数
```

3. 代码实践：Python 示例

为了更好地理解和应用 Redis Pub/Sub，我们通过 Python 代码示例来演示发布者和订阅者的实现。

环境准备:

确保已安装 Redis 服务器并运行。
安装 Redis Python 客户端库 redis-py:
```
pip install redis
```

示例代码:

publisher.py (发布者)


import redis
import time
import random
# Redis 连接配置
REDIS_HOST = 'localhost'
REDIS_PORT = 6379
REDIS_DB = 0
# 频道名称
CHANNEL_NAME = 'news.sports'
def publish_message(channel, message):
    """发布消息到指定频道"""
    r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
    publish_count = r.publish(channel, message)
    print(f"Published message to channel '{channel}': '{message}', Subscribers received: {publish_count}")
if __name__ == '__main__':
    sports_news = [
        "Breaking news: Team A wins championship!",
        "Exciting match: Team B defeats Team C in overtime.",
        "Player X sets new scoring record.",
        "Coach Y announces retirement at the end of the season."
    ]
    while True:
        news_item = random.choice(sports_news)
        publish_message(CHANNEL_NAME, news_item)
        time.sleep(random.uniform(1, 3)) # 随机间隔发布消息

subscriber.py (订阅者 - 频道订阅)


import redis
# Redis 连接配置
REDIS_HOST = 'localhost'
REDIS_PORT = 6379
REDIS_DB = 0
# 频道名称
CHANNEL_NAME = 'news.sports'
def subscribe_channel(channel):
    """订阅指定频道并接收消息"""
    r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
    pubsub = r.pubsub()
    pubsub.subscribe(channel)
    print(f"Subscribed to channel '{channel}'. Waiting for messages...")
    for message in pubsub.listen():
        if message['type'] == 'message':
            channel_name = message['channel'].decode('utf-8')
            message_data = message['data'].decode('utf-8')
            print(f"Received message from channel '{channel_name}': '{message_data}'")
if __name__ == '__main__':
    subscribe_channel(CHANNEL_NAME)

pattern_subscriber.py (订阅者 - 模式订阅)


import redis
# Redis 连接配置
REDIS_HOST = 'localhost'
REDIS_PORT = 6379
REDIS_DB = 0
# 模式
PATTERN = 'news.*'
def subscribe_pattern(pattern):
    """订阅指定模式并接收消息"""
    r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
    pubsub = r.pubsub()
    pubsub.psubscribe(pattern)
    print(f"Subscribed to pattern '{pattern}'. Waiting for messages...")
    for message in pubsub.listen():
        if message['type'] == 'pmessage':
            pattern_name = message['pattern'].decode('utf-8')
            channel_name = message['channel'].decode('utf-8')
            message_data = message['data'].decode('utf-8')
            print(f"Received message from pattern '{pattern_name}', channel '{channel_name}': '{message_data}'")
if __name__ == '__main__':
    subscribe_pattern(PATTERN)

运行示例:

启动订阅者:

打开一个终端窗口，运行 subscriber.py 或 pattern_subscriber.py。例如：
```
python subscriber.py
```
订阅者程序会连接到 Redis 并等待消息。
启动发布者:

打开另一个终端窗口，运行 publisher.py。
```
python publisher.py
```
发布者程序会随机选择体育新闻并发布到 news.sports 频道。

观察输出:

在订阅者终端窗口中，你将看到接收到的消息，类似于：


Subscribed to channel 'news.sports'. Waiting for messages...
Received message from channel 'news.sports': 'Breaking news: Team A wins championship!'
Received message from channel 'news.sports': 'Player X sets new scoring record.'
...

在发布者终端窗口中，你将看到发布消息的确认信息，以及接收到消息的订阅者数量。

代码详解:

publisher.py:
- 使用 redis.Redis() 创建 Redis 连接对象。
- publish_message() 函数使用 r.publish(channel, message) 发布消息。
- 主循环中，随机选择一条体育新闻，调用 publish_message() 发布，并随机等待一段时间。
subscriber.py 和 pattern_subscriber.py:
- 使用 redis.Redis() 创建 Redis 连接对象。
- r.pubsub() 创建 Pub/Sub 对象。
- pubsub.subscribe(channel) 或 pubsub.psubscribe(pattern) 订阅频道或模式。
- pubsub.listen() 进入消息监听循环。
- 循环遍历 pubsub.listen() 返回的消息，判断消息类型 (message 或 pmessage)，并提取频道名称和消息内容。
- 将接收到的消息打印到终端。

4. Redis Pub/Sub 的应用场景

Redis Pub/Sub 在各种场景中都有广泛的应用，以下是一些典型的例子：

实时聊天室: 每个聊天室可以作为一个频道。用户发送消息时，发布到聊天室频道，所有订阅该频道的用户即可实时接收到消息。
实时评论系统: 用户评论发布后，可以发布到评论频道，实时更新到页面上，无需轮询服务器。
新闻/事件推送: 新闻网站或应用可以将新闻事件分类到不同的频道（例如 news.sports, news.politics）。用户订阅感兴趣的频道，即可实时接收到相关新闻推送。
服务器监控和报警: 监控系统可以定期发布服务器状态信息到特定频道。报警系统订阅这些频道，一旦发现异常状态，立即发出报警通知。
分布式配置中心: 配置中心发布配置更新消息到频道。应用服务订阅配置频道，实时接收配置更新，动态调整自身行为。
微服务架构中的事件总线: 在微服务架构中，可以使用 Redis Pub/Sub 作为轻量级的事件总线，实现服务间的异步通信和事件驱动架构。
缓存失效通知: 当数据被修改时，发布缓存失效消息到频道。缓存服务订阅该频道，接收到消息后，主动失效相关缓存，保证数据一致性。

5. Redis Pub/Sub 的局限性与注意事项

虽然 Redis Pub/Sub 非常强大且易于使用，但也存在一些局限性，在使用时需要注意：

消息丢失: 无持久化是 Pub/Sub 的一个关键特性，但也意味着消息是易失的。如果订阅者在消息发布时离线或网络不稳定，则会丢失消息。Redis 不会存储未被消费的消息，也不会进行消息重传。对于需要可靠消息传递的场景，可能需要考虑使用 Redis Streams 或其他消息队列系统（如 RabbitMQ, Kafka）。
发布即推送: 消息一旦发布，Redis 服务器会立即尝试推送给所有订阅者。如果订阅者处理消息的速度跟不上发布速度，可能会导致消息堆积，甚至造成订阅者进程崩溃。需要考虑订阅者的消息处理能力，并可能需要进行限流或负载均衡。
单向通信: Pub/Sub 是一种单向通信模式，发布者只能发布消息，无法接收订阅者的反馈或响应。如果需要双向通信或请求-响应模式，需要使用其他机制。
客户端连接管理: 订阅模式会占用一个客户端连接。如果订阅者数量过多，可能会消耗大量的 Redis 连接资源。需要合理管理客户端连接，并考虑连接池等优化措施。
模式订阅的性能: 虽然模式订阅非常灵活，但在频道数量非常庞大的情况下，模式匹配可能会带来一定的性能开销。需要根据实际场景评估模式订阅的性能影响。

使用建议:

实时性优先，可靠性次要的场景: Pub/Sub 非常适合实时性要求高，但对消息丢失不敏感的场景，例如实时聊天、实时监控等。
轻量级消息通信: Pub/Sub 是一种轻量级的消息通信机制，易于部署和使用，适用于简单的消息广播场景。
结合其他机制增强可靠性: 如果需要更高的可靠性，可以考虑结合其他机制，例如：
- 客户端确认: 订阅者接收到消息后，向发布者发送确认消息，确保消息被成功处理。
- 消息重试机制: 在客户端实现消息重试逻辑，如果接收消息失败，可以尝试重新订阅或重新获取消息。
- 持久化存储: 将重要的消息同时写入持久化存储（例如数据库或日志文件），以便在消息丢失时进行恢复。
监控和调优: 监控 Redis Pub/Sub 的性能指标，例如消息发布速度、订阅者数量、消息延迟等。根据监控数据进行调优，例如调整频道设计、优化订阅者处理逻辑等。

6. 总结

Redis 发布订阅 (Pub/Sub) 是一种强大而灵活的消息通信模式，它为构建实时性应用提供了便捷的解决方案。通过本文的详细讲解和代码实践，相信你已经对 Redis Pub/Sub 的核心概念、命令操作、应用场景和局限性有了深入的理解。在实际应用中，需要根据具体需求权衡 Pub/Sub 的优缺点，并结合其他技术手段，构建稳定可靠的实时系统。

希望本文能够帮助你更好地掌握和应用 Redis 高级特性 - 发布订阅，提升你的技术能力。

作者与出处

原作者：灏天文库

来源：灏天文库

原文： https://aiknowledge.cn/collection/172-redis%E7%BC%93%E5%AD%98%E6%95%B0%E6%8D%AE%E5%BA%93%E5%AE%9E%E6%88%98%E6%95%99%E7%A8%8B

整理：灏天文库整理

由灏天文库平台收录，内容或由平台用户上传，仅供学习交流