使用 Redis Pub/Sub 构建消息系统 – wiki词典

作者： /

使用 Redis Pub/Sub 构建消息系统

在现代分布式系统中，消息系统扮演着至关重要的角色，用于实现服务间的解耦、异步通信和实时数据传输。Redis Pub/Sub（发布/订阅）机制作为一种轻量级且高性能的消息传递模式，因其简单、高效的特点，常被用于构建实时应用程序、消息广播和事件驱动架构。

什么是 Redis Pub/Sub？

Redis Pub/Sub 是一种一对多的消息传递模式。其核心思想是，消息的发送者（发布者）不会直接将消息发送给特定的接收者，而是将消息发布到一个或多个“通道”（Channel）中。消息的接收者（订阅者）则通过订阅这些通道来接收消息。发布者和订阅者之间彼此独立，互不感知对方的存在。

主要组成部分：

  1. 发布者 (Publisher): 负责创建消息并将其发送到指定的通道。发布者不知道也不关心是否有订阅者正在监听这些消息。
  2. 订阅者 (Subscriber): 负责监听一个或多个通道。一旦有消息发布到其订阅的通道，订阅者就会实时接收到该消息。订阅者也不知道消息是由哪个发布者发送的。
  3. 通道 (Channel): 是 Redis Pub/Sub 中消息传递的媒介。它是一个逻辑概念，由一个唯一的名称标识。发布者向通道发布消息，订阅者从通道接收消息。

Redis Pub/Sub 的工作原理

当发布者使用 PUBLISH channel_name message 命令将消息发送到某个通道时，Redis 服务器会将该消息推送给所有当前订阅了 channel_name 的客户端。这是一个实时的过程，消息不会在 Redis 服务器中持久化。

关键特性与优势

  • 解耦性 (Decoupling): 发布者和订阅者之间不存在直接依赖，它们可以独立开发、部署和扩展。这大大提高了系统的灵活性和可维护性。
  • 实时性 (Real-time): Redis 作为内存数据库，拥有极高的读写性能。Pub/Sub 机制能够确保消息以极低的延迟在发布者和订阅者之间传递，非常适合实时通知、聊天应用等场景。
  • 广播能力 (Broadcast): 一条消息可以同时发送给所有订阅者，非常适合需要将信息广播给多个客户端的场景。
  • 简单易用: Redis Pub/Sub 的 API 设计简洁直观，易于上手和集成。

局限性

尽管 Redis Pub/Sub 具有诸多优势，但它也存在一些局限性，理解这些局限性对于正确选择消息系统至关重要：

  • “即发即忘” (Fire-and-forget): Redis Pub/Sub 不提供任何形式的消息持久化。如果消息发布时没有活跃的订阅者在线，该消息将永远丢失，无法被后续连接的订阅者接收。
  • 无消息确认机制: 发布者无法得知消息是否被订阅者成功接收。
  • 订阅者必须在线: 如果订阅者在消息发布时处于离线状态，它将错过在此期间发布的所有消息。

对于需要消息持久化、消息确认、更复杂的消息路由或处理离线订阅者的场景，可能需要考虑使用更专业的队列服务，如 RabbitMQ、Kafka 或 Redis Streams（Redis 5.0+ 引入的持久化消息队列）。

使用 Python 和 redis-py 构建示例消息系统

下面我们将通过一个简单的 Python 示例，演示如何使用 redis-py 库来构建一个基本的 Redis Pub/Sub 消息系统。

前提条件

  1. 安装 Redis 服务器: 确保您的系统上运行着 Redis 服务器。
  2. 安装 Python: 确保您的系统上安装了 Python 3。
  3. 安装 redis-py 库:
    bash
    pip install redis

1. 发布者 (publisher.py)

这个脚本会连接到 Redis，并每秒向名为 my_channel 的通道发布一条消息。

“`python

publisher.py

import redis
import time
import sys

连接到 Redis 服务器

默认连接到 localhost:6379

try:
r = redis.Redis(host=’localhost’, port=6379, db=0)
r.ping() # 尝试连接，如果失败会抛出异常
print(“成功连接到 Redis 服务器”)
except redis.exceptions.ConnectionError as e:
print(f”无法连接到 Redis 服务器: {e}”)
sys.exit(1)

channel_name = “my_channel”
message_count = 0

print(f”开始向通道 ‘{channel_name}’ 发布消息…”)

try:
while True:
message_count += 1
message = f”Hello from publisher! Message {message_count}”
# 发布消息到指定通道
r.publish(channel_name, message)
print(f”已发布: ‘{message}’ 到通道 ‘{channel_name}'”)
time.sleep(1) # 每秒发布一条消息
except KeyboardInterrupt:
print(“\n发布者已停止。”)
except Exception as e:
print(f”发布者发生错误: {e}”)

“`

2. 订阅者 (subscriber.py)

这个脚本会连接到 Redis，订阅 my_channel 通道，并实时打印所有接收到的消息。

“`python

subscriber.py

import redis
import sys

连接到 Redis 服务器

try:
r = redis.Redis(host=’localhost’, port=6379, db=0)
r.ping()
print(“成功连接到 Redis 服务器”)
except redis.exceptions.ConnectionError as e:
print(f”无法连接到 Redis 服务器: {e}”)
sys.exit(1)

channel_name = “my_channel”

创建一个 PubSub 对象

pubsub = r.pubsub()

订阅通道

pubsub.subscribe(channel_name)
print(f”已订阅通道 ‘{channel_name}’，等待消息…”)

try:
# 监听消息
# pubsub.listen() 是一个生成器，会阻塞直到收到消息
for message in pubsub.listen():
if message[‘type’] == ‘message’:
# Redis 返回的消息是字节类型，需要解码
decoded_message = message[‘data’].decode(‘utf-8’)
print(f”收到消息: ‘{decoded_message}'”)
except KeyboardInterrupt:
print(“\n订阅者已停止。”)
except Exception as e:
print(f”订阅者发生错误: {e}”)
finally:
# 在程序结束时取消订阅
pubsub.unsubscribe(channel_name)
print(f”已取消订阅通道 ‘{channel_name}’。”)

“`

如何运行示例

  1. 启动 Redis 服务器 (如果尚未运行，可以通过 redis-server 命令或 Docker 启动)。
  2. 打开两个独立的终端窗口。
  3. 在第一个终端中运行订阅者脚本：
    bash
    python subscriber.py
    您应该会看到类似如下的输出，表示订阅者正在等待消息：
    成功连接到 Redis 服务器
    已订阅通道 'my_channel'，等待消息...
  4. 在第二个终端中运行发布者脚本：
    bash
    python publisher.py
    发布者将开始每秒发送一条消息：
    成功连接到 Redis 服务器
    开始向通道 'my_channel' 发布消息...
    已发布: 'Hello from publisher! Message 1' 到通道 'my_channel'
    已发布: 'Hello from publisher! Message 2' 到通道 'my_channel'
    ...
  5. 观察订阅者终端：
    订阅者终端将实时接收并打印发布者发送的消息：
    收到消息: 'Hello from publisher! Message 1'
    收到消息: 'Hello from publisher! Message 2'
    收到消息: 'Hello from publisher! Message 3'
    ...

总结

Redis Pub/Sub 提供了一种强大而简单的机制来实现实时、解耦的消息传递。它非常适合需要高吞吐量、低延迟消息广播的场景，例如实时通知、聊天室、仪表盘更新等。然而，由于其“即发即忘”的特性，对于需要消息可靠性、持久化或复杂消息队列功能的场景，开发者应考虑结合使用 Redis Streams 或其他更专业的 MQ 解决方案。理解其优缺点并结合实际需求，能够帮助我们更好地利用 Redis Pub/Sub 构建健壮高效的分布式系统。

滚动至顶部