掌握 Redis 发布订阅 (Pub/Sub) – wiki基地

作者： /

深度探索 Redis 发布订阅 (Pub/Sub): 原理、实践与局限

在构建现代分布式系统时，服务间的解耦和实时通信是一个常见的需求。传统的方式可能是通过 HTTP 请求或 RPC 调用，但这种方式通常是同步的、紧耦合的。而消息队列和发布订阅模式则提供了一种异步、解耦的通信方式。

Redis，作为一款高性能的内存数据结构存储，不仅提供了键值存储、列表、集合、有序集合、哈希等丰富的数据结构，还内置了对发布订阅 (Publish/Subscribe, Pub/Sub) 模式的支持。尽管 Redis 的 Pub/Sub 实现相对简单，不具备传统消息队列（如 Kafka、RabbitMQ）的持久化和高级特性，但其高性能和易用性使其在某些特定场景下成为一个非常有吸引力的选择。

本文将深入探讨 Redis 的 Pub/Sub 机制，从其基本原理、核心命令、实际应用场景，到其重要的局限性和与其他消息机制的比较，帮助读者全面掌握 Redis Pub/Sub 的精髓。

第一部分：理解发布订阅 (Pub/Sub) 模型

发布订阅是一种消息范式，其中消息的发送者（发布者，Publisher）不会直接将消息发送给特定的接收者（订阅者，Subscriber）。相反，发布者将消息发布到特定的类别或通道（Channel），而订阅者可以表达对一个或多个通道的兴趣，只有订阅了某个通道的订阅者才能接收到发布到该通道的消息。

这种模型的核心在于：

  1. 解耦 (Decoupling): 发布者和订阅者彼此不需要知道对方的存在。发布者只知道将消息发送到哪个通道，订阅者只知道从哪个通道接收消息。这种解耦使得系统更容易扩展和维护。
  2. 异步 (Asynchronous): 发布消息的操作通常是即时完成的，发布者不需要等待订阅者的响应。消息的消费由订阅者独立完成。
  3. 多播 (Multicast): 一条消息可以被多个订阅者同时接收。这是与点对点（Point-to-Point）消息队列（如只被一个消费者接收的 FIFO 队列）最主要的区别之一。

在 Redis Pub/Sub 模型中：

  • 发布者 (Publisher): 发送消息到某个通道的客户端。
  • 订阅者 (Subscriber): 接收来自某个通道消息的客户端。
  • 通道 (Channel): 消息的类别或主题，发布者将消息发送到通道，订阅者从通道接收消息。

第二部分：Redis Pub/Sub 的工作原理

Redis 的 Pub/Sub 实现非常直接和高效。它基于一个简单的内存数据结构来维护通道和订阅者之间的映射关系。

  1. 订阅过程: 当一个客户端执行 SUBSCRIBE channel_name 命令时，Redis 会在内部维护一个数据结构（例如一个哈希表），将 channel_name 与该客户端的连接关联起来。如果这是该通道的第一个订阅者，Redis 会创建或查找对应的通道条目。
  2. 发布过程: 当另一个客户端执行 PUBLISH channel_name message 命令时，Redis 会查找与 channel_name 相关联的所有订阅者连接。
  3. 消息投递: Redis 遍历所有订阅了 channel_name 的客户端连接，并将 message 发送到每个连接。

关键特点:

  • 无消息持久化: Redis Pub/Sub 的一个重要特性是它不存储消息。消息一旦被发布，Redis 会立即尝试将其发送给所有当前在线且订阅了该通道的客户端。如果某个客户端在消息发布时没有订阅或处于离线状态，那么它将永远无法收到这条消息。这是一种“火候即焚”（fire-and-forget）或“至多一次”（at-most-once）的投递模式。
  • 内存操作: 整个过程都在内存中进行，这使得 Redis Pub/Sub 的发布和投递速度非常快，延迟很低。
  • 阻塞式订阅: 客户端一旦执行 SUBSCRIBE 或 PSUBSCRIBE 命令，该连接就会进入订阅模式。在这个模式下，该客户端连接将不再能执行除订阅、退订、退出等少数命令之外的任何其他 Redis 命令。它只会接收到两种类型的回复：订阅确认信息（如 SUBSCRIBED）和接收到的消息（格式通常是 message, channel, content）。

第三部分：核心命令详解与示例

Redis Pub/Sub 功能主要通过以下几个核心命令实现：

  1. PUBLISH channel message

    • 功能: 将 message 发布到指定的 channel。
    • 返回值: 接收到这条消息的订阅者数量。注意，这只计算当前活跃并订阅了该通道的客户端，不包括使用模式订阅的客户端。
    • 示例:
      redis
      PUBLISH news "Hello subscribers!"
      (integer) 2 # 表示有2个客户端收到了这条消息
      在订阅了 news 通道的客户端会收到如下回复：
      1) "message"
      2) "news"
      3) "Hello subscribers!"

  2. SUBSCRIBE channel [channel ...]

    • 功能: 订阅指定的一个或多个通道。
    • 返回值: 对于每个成功的订阅，客户端会收到一条回复，格式为 (integer) "subscribe" (string) channel_name (integer) number_of_subscribed_channels。一旦进入订阅模式，客户端连接将阻塞，直到收到消息或执行退订命令。

    • 示例:
      客户端 A 执行：
      redis
      SUBSCRIBE channel1 channel2
      客户端 A 会收到：
      “`
      1) “subscribe” # 订阅类型
      2) “channel1” # 订阅成功的通道名
      3) (integer) 1 # 当前连接订阅的通道总数

      1) “subscribe”
      2) “channel2”
      3) (integer) 2
      “`
      此时，客户端 A 进入订阅模式。

      当有消息发布到 channel1 时 (例如，另一个客户端执行 PUBLISH channel1 "message for channel1"):
      客户端 A 会收到：
      1) "message" # 消息类型
      2) "channel1" # 接收到消息的通道名
      3) "message for channel1" # 消息内容

  3. UNSUBSCRIBE [channel [channel ...]]

    • 功能: 退订指定的一个或多个通道。如果不指定通道，则退订当前连接订阅的所有通道。当所有通道都被退订后，客户端将退出订阅模式。
    • 返回值: 对于每个成功的退订，客户端会收到一条回复，格式为 (integer) "unsubscribe" (string) channel_name (integer) number_of_subscribed_channels。当 number_of_subscribed_channels 变为 0 时，表示客户端已退出订阅模式。
    • 示例:
      客户端 A 在订阅 channel1 和 channel2 后，执行：
      redis
      UNSUBSCRIBE channel1
      客户端 A 会收到：
      1) "unsubscribe"
      2) "channel1"
      3) (integer) 1 # 当前还订阅着 channel2
      然后执行：
      redis
      UNSUBSCRIBE
      客户端 A 会收到：
      1) "unsubscribe"
      2) "channel2"
      3) (integer) 0 # 所有通道已退订，退出订阅模式

  4. PSUBSCRIBE pattern [pattern ...]

    • 功能: 订阅与给定模式匹配的通道。模式使用通配符 * (匹配任意数量的任意字符) 和 ? (匹配单个任意字符)。
    • 返回值: 对于每个成功的模式订阅，客户端会收到一条回复，格式为 (integer) "psubscribe" (string) pattern (integer) number_of_subscribed_patterns_and_channels。客户端进入订阅模式。

    • 示例:
      客户端 B 执行：
      redis
      PSUBSCRIBE news.* chat.?
      客户端 B 会收到：
      “`
      1) “psubscribe”
      2) “news.*”
      3) (integer) 1

      1) “psubscribe”
      2) “chat.?”
      3) (integer) 2
      “`
      此时，客户端 B 进入订阅模式。

      如果发布者发布消息到 news.sports 或 news.technology：
      redis
      PUBLISH news.sports "Sports news update"
      PUBLISH news.technology "Tech news update"
      客户端 B 会收到：
      “`
      1) “pmessage” # 模式匹配消息类型
      2) “news.*” # 匹配到的模式
      3) “news.sports” # 实际接收到消息的通道名
      4) “Sports news update” # 消息内容

      1) “pmessage”
      2) “news.*”
      3) “news.technology”
      4) “Tech news update”
      “`

      如果发布者发布消息到 chat.a 或 chat.1:
      redis
      PUBLISH chat.a "Message for chat a"
      PUBLISH chat.1 "Message for chat 1"
      客户端 B 会收到：
      “`
      1) “pmessage”
      2) “chat.?”
      3) “chat.a”
      4) “Message for chat a”

      1) “pmessage”
      2) “chat.?”
      3) “chat.1”
      4) “Message for chat 1”
      ``
      注意，使用模式订阅时，客户端收到消息的格式是      (“pmessage”, pattern, channel, message)，而不是(“message”, channel, message)`。

  5. PUNSUBSCRIBE [pattern [pattern ...]]

    • 功能: 退订指定的一个或多个模式订阅。如果不指定模式，则退订当前连接订阅的所有模式。当所有通道和模式都被退订后，客户端将退出订阅模式。
    • 返回值: 对于每个成功的退订，客户端会收到回复，格式为 (integer) "punsubscribe" (string) pattern (integer) number_of_subscribed_patterns_and_channels。
    • 示例:
      客户端 B 在订阅了 news.* 和 chat.? 后，执行：
      redis
      PUNSUBSCRIBE news.*
      客户端 B 会收到：
      1) "punsubscribe"
      2) "news.*"
      3) (integer) 1 # 还订阅着 chat.?
      然后执行：
      redis
      UNSUBSCRIBE # 退订所有普通通道订阅
      PUNSUBSCRIBE # 退订所有模式订阅
      客户端 B 会收到：
      1) "punsubscribe"
      2) "chat.?"
      3) (integer) 0 # 所有模式和通道都已退订，退出订阅模式

  6. PUBSUB subcommand [argument [argument ...]]

    • 功能: 提供对 Pub/Sub 子系统的内省能力。
    • 常用子命令:
      • PUBSUB CHANNELS [pattern]: 列出当前活跃的通道。如果指定模式，则只列出匹配的通道。
      • PUBSUB NUMSUB [channel [channel ...]]: 返回指定通道的订阅者数量。
      • PUBSUB NUMPAT: 返回通过 PSUBSCRIBE 订阅模式的客户端数量（注意不是模式的数量，而是连接的数量）。

    • 示例:
      “`redis
      # 假设 channel1, channel2, news.sports, chat.a 有活跃订阅者
      PUBSUB CHANNELS
      1) “channel1”
      2) “channel2”
      3) “news.sports”
      4) “chat.a”

      PUBSUB CHANNELS news.*
      1) “news.sports”

      PUBSUB NUMSUB channel1 channel2 news.sports
      1) “channel1”
      2) (integer) 1 # channel1 有1个订阅者
      3) “channel2”
      4) (integer) 2 # channel2 有2个订阅者
      5) “news.sports”
      6) (integer) 1 # news.sports 有1个订阅者 (注意这里只统计直接订阅，不统计模式订阅)

      PUBSUB NUMPAT
      (integer) 3 # 假设有3个客户端使用了 PSUBSCRIBE 订阅了某个模式
      “`

第四部分：Redis Pub/Sub 的典型应用场景

基于其高性能、实时性和多播特性，Redis Pub/Sub 在许多场景中都能发挥作用：

  1. 实时聊天系统: 这是最经典的 Pub/Sub 应用场景之一。每个聊天室可以对应一个 Redis 通道。当用户加入聊天室时，客户端订阅该聊天室对应的通道。当用户发送消息时，服务器将消息发布到该聊天室通道。所有订阅了该通道的用户客户端都会实时收到消息。
  2. 实时数据更新/通知: 例如，股票行情、体育赛事比分、用户在线状态变化、订单状态更新等。后端服务可以将这些实时变化的数据发布到相应的通道，前端客户端或其他服务订阅这些通道，即可实时接收更新。仪表盘应用也可以通过订阅通道来实时显示关键指标变化。
  3. 缓存失效通知: 在分布式缓存系统中，当某个服务更新了数据并导致缓存需要失效时，它可以向一个特定的缓存失效通道发布消息，包含失效的键信息。其他所有依赖这个缓存的服务都订阅这个通道，接收到消息后将本地缓存对应的键设置为失效。
  4. 简单的事件总线/系统通知: 在微服务架构中，服务之间可能需要对某些事件做出反应。Pub/Sub 可以作为一个简单的事件总线。例如，用户注册服务在完成用户注册后，可以发布一个“user.registered”事件到相应的通道。其他服务（如邮件发送服务、积分服务）订阅这个通道，接收到事件后执行各自的后续操作。
  5. 任务队列（非关键任务）: 虽然 Redis Lists 更常用于构建任务队列（生产者/消费者模式），但对于一些非关键、可以丢失、需要多方监听的任务，Pub/Sub 也可以使用。例如，一个服务发布一个“生成报告”的任务到通道，多个工作节点订阅该通道，每个节点收到消息后都可以尝试生成报告（需要额外机制避免重复）。但请注意，由于 Pub/Sub 不保证消息送达，这种方式不适合需要确保任务被处理的场景。

第五部分：在不同编程语言中使用 Redis Pub/Sub

几乎所有的主流编程语言都有成熟的 Redis 客户端库，这些库都支持 Pub/Sub 功能。使用方式大同小异：

  1. 获取 Redis 连接: 创建一个 Redis 客户端连接。
  2. 进入订阅模式: 使用客户端提供的 subscribe 或 psubscribe 方法订阅通道或模式。重要提示: 执行订阅后，该连接会进入阻塞模式，专用于接收消息。不能使用同一个连接执行其他 Redis 命令（如 GET, SET 等）。通常需要为 Pub/Sub 操作创建一个专门的连接。
  3. 监听消息: 订阅方法通常会返回一个 Pub/Sub 对象或进入一个循环，持续监听来自 Redis 的消息。当收到消息时，会触发一个回调函数或生成一个事件。
  4. 发布消息: 使用另一个独立的 Redis 连接，调用 publish 方法发布消息到指定通道。

概念性代码示例 (以 Python 使用 redis-py 库为例):

“`python

发布者 (Publisher) 代码示例 – 在一个独立的连接上

import redis

r = redis.Redis(decode_responses=True) # decode_responses=True 直接返回字符串

发布消息

num_subscribers = r.publish(‘my_channel’, ‘Hello, subscribers!’)
print(f”Published message to ‘my_channel’. Received by {num_subscribers} subscribers.”)

订阅者 (Subscriber) 代码示例 – 需要在另一个独立的连接上，通常运行在循环或线程中

import redis
import time

r_sub = redis.Redis(decode_responses=True)
p = r_sub.pubsub()

订阅通道

p.subscribe(‘my_channel’)
p.subscribe(‘another_channel’)
p.psubscribe(‘news.*’)

print(“Subscribed to channels and patterns. Listening for messages…”)

进入监听循环

p.listen() 会阻塞直到收到消息

for message in p.listen():
# message 格式示例:
# {‘type’: ‘subscribe’, ‘channel’: ‘my_channel’, ‘data’: 1}
# {‘type’: ‘message’, ‘channel’: ‘my_channel’, ‘data’: ‘Hello, subscribers!’}
# {‘type’: ‘psubscribe’, ‘pattern’: ‘news.‘, ‘channel’: b’news.‘, ‘data’: 1}
# {‘type’: ‘pmessage’, ‘pattern’: ‘news.*’, ‘channel’: ‘news.sports’, ‘data’: ‘Sports update’}

if message['type'] == 'message':
    print(f"Received message: Channel='{message['channel']}', Data='{message['data']}'")
elif message['type'] == 'pmessage':
    print(f"Received pattern message: Pattern='{message['pattern']}', Channel='{message['channel']}', Data='{message['data']}'")
elif message['type'] == 'subscribe':
    print(f"Subscribed to: Channel='{message['channel']}', Total='{message['data']}'")
elif message['type'] == 'psubscribe':
     print(f"Subscribed to pattern: Pattern='{message['pattern']}', Total='{message['data']}'")
# 处理其他类型如 unsubscribe, punsubscribe 等

“`

在实际应用中，订阅者通常会将监听循环放在一个单独的线程或进程中，以避免阻塞主程序。

第六部分：Redis Pub/Sub 的局限性与注意事项

虽然 Redis Pub/Sub 简单高效，但它有几个重要的局限性，需要在使用前充分理解：

  1. 无消息持久化 (No Message Persistence): 这是最关键的局限。Redis Pub/Sub 不会存储已发布的消息。如果消息发布时没有客户端订阅，或者订阅者客户端由于网络问题、程序崩溃等原因暂时离线，那么这些客户端将永远丢失在这期间发布的所有消息。一旦消息从发布者发送到 Redis，Redis 会立即尝试投递，投递完成后（或因无订阅者而无法投递）消息即被“忘记”。
  2. 无送达保证 (At-Most-Once Delivery): 由于无持久化，Redis 只能做到“至多一次”的送达。它会尽力将消息投递给当前在线的订阅者，但无法保证消息一定会被接收。如果网络不稳定或订阅者处理消息过慢导致缓冲溢出，消息可能会丢失。
  3. 无消费者组 (No Consumer Groups): 传统的消息队列（如 Kafka, RabbitMQ）支持消费者组的概念，允许多个消费者共同消费同一个队列或主题中的消息，且一条消息只会被组内的一个消费者处理。Redis Pub/Sub 是纯粹的多播，所有订阅了同一通道的客户端都会收到消息的 副本。无法实现竞争消费（Work Queue）模式。
  4. 阻塞式订阅 (Blocking Nature): 如前所述，执行 SUBSCRIBE 或 PSUBSCRIBE 的连接会进入阻塞模式，不能用于执行其他 Redis 命令。这要求应用程序为 Pub/Sub 创建单独的连接。
  5. 慢订阅者问题 (Slow Subscriber Problem): 如果一个订阅者处理消息的速度跟不上发布的速度，Redis 会在服务器端为该订阅者维护一个输出缓冲区。如果这个缓冲区持续增长并达到 Redis 配置的限制（client-output-buffer-limit pubsub <hard limit> <soft limit> <soft seconds>），Redis 会强制断开这个慢订阅者的连接。这样做是为了保护 Redis 服务器自身的内存和性能，避免被少数慢客户端拖垮。断开连接后，慢订阅者将丢失连接断开期间发布的所有消息。
  6. 扩展性限制 (Scalability for Subscribers): 当通道上的消息量非常大，并且有大量的订阅者时，Redis 服务器需要向每个订阅者发送每条消息的副本。这会消耗大量的网络带宽和 CPU 资源。对于极高吞吐量的 Pub/Sub 场景，可能需要考虑更专业的分布式消息队列系统。

第七部分：Pub/Sub 与 Redis 其他消息机制的比较

除了 Pub/Sub，Redis 还提供了其他可以实现消息通信的数据结构和命令。理解它们之间的区别有助于选择合适的工具。

  1. Pub/Sub vs. Redis Lists (LPUSH/BRPOP):

    • 模型: Lists 常用于构建简单的点对点队列或工作队列。生产者使用 LPUSH 将任务放入列表头部，消费者使用 BRPOP (阻塞式弹出) 从列表尾部获取任务。一条消息（列表元素）通常只会被一个消费者消费。
    • 持久性: Lists 数据是持久化的 (除非 Redis 关闭时未配置 RDB/AOF 持久化)。即使消费者离线，生产者发布的消息也会保留在列表中，待消费者上线后可以继续消费。
    • 送达保证: 提供了更好的送达保证（至少一次，取决于消费者是否成功处理并删除）。
    • 使用场景: 简单的任务队列、工作队列、异步处理。
    • Pub/Sub 优势: 多播、实时性高、实现简单（发布者和订阅者只需知道通道名），对发布者无压力（不关心有多少订阅者，也不需要存储消息）。
    • Lists 优势: 消息持久化、送达保证、竞争消费。

  2. Pub/Sub vs. Redis Streams (XADD/XREAD/XREADGROUP):

    • 模型: Redis Streams 是 Redis 5.0 引入的更高级的消息队列数据结构。它是一个持久化的、只追加的日志结构，支持多个生产者和多个消费者，并引入了消费者组概念。
    • 持久性: 消息默认是持久化的。
    • 送达保证: 提供了更强的送达保证（至少一次，结合 XACK 可以实现恰好一次）。支持消息确认（ACK）。
    • 消费者模型: 支持消费者组，可以在组内实现竞争消费，也可以独立消费（类似 Pub/Sub 的多播，但不直接是 Pub/Sub）。支持从任意位置读取历史消息。
    • 复杂度: Streams 的概念和命令比 Pub/Sub 更复杂。
    • 使用场景: 需要消息持久化、送达保证、消费者组、处理历史消息、更复杂的分布式消息队列场景。
    • Pub/Sub 优势: 极其简单、纯粹的实时多播、低延迟、高性能（尤其在消息量大但订阅者不极多的情况下）。
    • Streams 优势: 持久化、送达保证、消费者组、可回溯、更适合构建健壮的消息系统。

总结选择:

  • 选择 Pub/Sub: 如果你的需求是：纯粹的实时多播通知，消息丢失可以接受，不需要持久化，追求极低的延迟和高吞吐（发布端），订阅者数量在可控范围。典型的如实时聊天、非关键的系统通知、缓存失效通知。
  • 选择 Lists: 如果你的需求是：简单的点对点任务队列或工作队列，消息需要持久化但不需要复杂的消费者组和回溯功能。
  • 选择 Streams: 如果你的需求是：需要可靠的消息队列、消息必须持久化且不能丢失、需要复杂的消费者组管理、需要处理消息积压或回溯历史消息。

第八部分：使用 Redis Pub/Sub 的最佳实践

基于前文的分析，使用 Redis Pub/Sub 时可以遵循以下最佳实践：

  1. 明确消息丢失可接受: 仅将 Pub/Sub 用于那些即使消息丢失也不会对核心业务产生严重影响的场景。对于支付、订单、关键通知等需要严格送达保证的场景，请考虑使用 Lists（配合持久化和重试机制）或 Streams。
  2. 为 Pub/Sub 创建专用连接: 订阅客户端必须使用一个专门的连接进入订阅模式，不要尝试用同一个连接执行其他 Redis 命令。
  3. 处理客户端断开连接和重连: 订阅者客户端需要健壮地处理与 Redis 服务器的断开连接情况（网络闪断、服务器重启、慢订阅者被断开等）。在连接断开后，客户端应实现指数退避等策略尝试重新连接，并重新执行 SUBSCRIBE 或 PSUBSCRIBE 命令。请注意，重连并重新订阅后，断开期间的所有消息都会丢失。
  4. 监控慢订阅者: 关注 Redis 的 client-output-buffer-limit pubsub 配置。如果业务场景可能出现慢订阅者，考虑调整这些限制，或者更积极地监控客户端连接情况，识别并处理慢客户端，避免影响整个 Redis 服务器。
  5. 保持消息简洁: 尽管 Redis Pub/Sub 可以传输任意二进制数据，但为了性能和带宽考虑，消息内容应尽量简洁，只包含必要的信息（例如，只需发送变化的键或事件 ID，而不是完整的数据对象）。接收者可以根据消息中的 ID 再去查询完整数据。
  6. 利用模式订阅进行分组: 如果通道数量庞大且有命名规律，可以利用 PSUBSCRIBE 进行模式订阅，简化管理。
  7. 监控 Pub/Sub 活跃度: 使用 PUBSUB 命令或 Redis 监控工具，定期检查活跃通道和订阅者数量，了解 Pub/Sub 系统的运行状态。

结论

Redis 的发布订阅 (Pub/Sub) 功能以其简单、高性能和多播特性，为许多实时应用场景提供了强大的支持。它使得发布者和订阅者能够高度解耦地进行异步通信，非常适合实现实时通知、消息推送、缓存失效等非关键但对实时性要求较高的功能。

然而，其无消息持久化和无送达保证是两个核心局限，这意味着它不适合用于需要可靠消息传递或处理消息积压的关键业务系统。在这种情况下，Redis Lists（用于简单队列）或更强大的 Redis Streams（用于需要持久化、送达保证和消费者组的场景）是更合适的选择，或者考虑 RabbitMQ、Kafka 等专业的消息中间件。

全面理解 Redis Pub/Sub 的工作原理、掌握其核心命令、知晓其优势与局限，并结合实际应用场景进行权衡，才能在正确的地方高效地利用这一强大的 Redis 特性。通过遵循最佳实践，可以最大化地发挥 Redis Pub/Sub 的价值，构建更具响应性和可扩展性的分布式系统。

发表评论

您的邮箱地址不会被公开。 必填项已用 * 标注

滚动至顶部