各位观众，各位朋友，大家好！我是你们的老朋友，今天咱们来聊聊 Redis Cluster 里的 Pub/Sub，也就是发布/订阅功能。这可不是什么高深莫测的黑魔法，而是 Redis Cluster 为了让消息在各个节点之间愉快地传递而设计的一套机制。
Pub/Sub：消息的“广播喇叭”
想象一下，你是个电台DJ，手里拿着个大喇叭，要向全世界广播你的音乐。Pub/Sub 就像这个大喇叭，Publisher（发布者）负责把消息“喊”出去，Subscriber（订阅者）则负责“听”喇叭里的内容。
在 Redis 的世界里，Publisher 把消息发布到一个特定的 Channel（频道）上，所有订阅了这个 Channel 的 Subscriber 就能收到这条消息。简单来说，就是“你喊一声，大家都能听到”。
Redis Cluster 里的 Pub/Sub：更复杂的“喇叭系统”
现在，我们把场景升级一下。假设你不是一个简单的电台，而是一个大型广播集团，在全国各地都有分台。每个分台都有自己的喇叭，需要协同工作才能把消息广播到全国。这就是 Redis Cluster 里的 Pub/Sub。
在 Redis Cluster 中，数据被分散存储在不同的节点上。这意味着，Publisher 和 Subscriber 可能位于不同的节点上。那么，消息是如何在各个节点之间传递的呢？这就涉及到 Redis Cluster 的 Pub/Sub 实现机制。
Pub/Sub 在 Redis Cluster 中的工作原理
Redis Cluster 的 Pub/Sub 并没有像数据分片那样，把 Channel 分散到不同的节点上。相反，它采用了一种“广播”的方式：
- 消息发布： 当 Publisher 发布消息时，消息会被发送到 Publisher 连接的节点。
- 节点广播： 收到消息的节点会将消息广播到 Cluster 中的所有节点。注意，是所有节点！
- 消息传递： 每个节点收到消息后，会检查是否有 Subscriber 订阅了该 Channel。如果有，就把消息传递给对应的 Subscriber。
是不是有点像“烽火狼烟”？一个节点发现有消息，就点燃烽火，通知所有节点。
代码示例：Pub/Sub 的基本操作
废话不多说，咱们直接上代码，用 Python 版本的 Redis 客户端 redis-py 来演示一下 Pub/Sub 的基本操作。
首先，安装 redis-py:
pip install redis然后，连接到 Redis Cluster（假设已经搭建好了集群，并且可以访问）。这里为了方便演示，假设集群只有一个主节点，但是代码的逻辑在多节点集群下仍然适用。
import redis
# 连接到 Redis Cluster (这里假设只有一个节点)
redis_cluster = redis.Redis(host='localhost', port=6379)
# 创建 PubSub 对象
pubsub = redis_cluster.pubsub()1. 订阅 Channel
Subscriber 需要先订阅 Channel 才能接收消息。
def message_handler(message):
print(f"Received message: {message['data'].decode()}")
pubsub.subscribe(**{'my_channel': message_handler}) # 订阅 'my_channel' 频道，指定消息处理函数
# 启动监听，等待消息 (这个操作会阻塞主线程)
# pubsub.run_in_thread(sleep_time=0.01) #推荐使用线程方式
thread = pubsub.run_in_thread(sleep_time=0.01) # 创建一个线程来处理消息这里，我们定义了一个 message_handler 函数，用于处理接收到的消息。然后，我们使用 pubsub.subscribe() 方法订阅了名为 my_channel 的 Channel，并将 message_handler 函数作为回调函数。 pubsub.run_in_thread() 在一个独立的线程中运行 Pub/Sub 监听器，这样主线程不会被阻塞。
2. 发布消息
Publisher 可以向 Channel 发布消息。
# 发布消息
redis_cluster.publish('my_channel', 'Hello, Redis Cluster!') # 向 'my_channel' 频道发布消息
redis_cluster.publish('my_channel', 'This is a test message.') # 向 'my_channel' 频道发布消息这里，我们使用 redis_cluster.publish() 方法向 my_channel Channel 发布了两条消息。
3. 取消订阅 Channel
Subscriber 可以随时取消订阅 Channel。
pubsub.unsubscribe('my_channel') # 取消订阅 'my_channel' 频道
thread.stop() # 停止线程这里，我们使用 pubsub.unsubscribe() 方法取消了对 my_channel Channel 的订阅。同时，也停止了之前开启的线程，避免资源浪费。
完整代码示例
import redis
import time
# 连接到 Redis Cluster (这里假设只有一个节点)
redis_cluster = redis.Redis(host='localhost', port=6379)
# 创建 PubSub 对象
pubsub = redis_cluster.pubsub()
def message_handler(message):
print(f"Received message: {message['data'].decode()}")
pubsub.subscribe(**{'my_channel': message_handler}) # 订阅 'my_channel' 频道，指定消息处理函数
thread = pubsub.run_in_thread(sleep_time=0.01) # 创建一个线程来处理消息
# 模拟 Publisher 发布消息
time.sleep(1) # 等待 Subscriber 启动
redis_cluster.publish('my_channel', 'Hello, Redis Cluster!') # 向 'my_channel' 频道发布消息
redis_cluster.publish('my_channel', 'This is a test message.') # 向 'my_channel' 频道发布消息
time.sleep(1) # 等待消息处理完成
pubsub.unsubscribe('my_channel') # 取消订阅 'my_channel' 频道
thread.stop() # 停止线程
print("Finished!")运行上面的代码，你应该能在控制台上看到 Subscriber 接收到的消息。
Pub/Sub 的局限性与注意事项
虽然 Pub/Sub 用起来很方便，但它也有一些局限性，需要我们注意：
- 无消息持久化： Pub/Sub 是一种“fire and forget”的模式。消息一旦发布，就不会被存储。如果 Subscriber 在消息发布时没有在线，就无法收到这条消息。这与消息队列（如 Redis Streams 或 RabbitMQ）不同，后者可以持久化消息，确保消息不会丢失。
- 广播风暴： 由于 Redis Cluster 会将消息广播到所有节点，如果消息量很大，可能会导致网络拥塞，影响集群的性能。这就是所谓的“广播风暴”。
- 无消息确认： Publisher 无法知道 Subscriber 是否成功接收了消息。如果 Subscriber 处理消息失败，Publisher 也无法收到通知。
- 模式订阅的限制: Redis Cluster 对模式订阅（使用通配符订阅多个Channel）的支持有限，性能可能较差，不建议在生产环境中使用。
如何解决 Pub/Sub 的局限性？
针对 Pub/Sub 的局限性，我们可以采取一些措施来缓解：
- 使用消息队列： 如果需要消息持久化和可靠性保证，应该使用消息队列，而不是 Pub/Sub。Redis Streams 是一个不错的选择，它提供了持久化、消息确认、消费者组等功能。
- 限制消息大小和频率： 尽量控制消息的大小和发布频率，避免产生广播风暴。
- 使用 Pipelining： Publisher 可以使用 Pipelining 技术批量发布消息，减少网络开销。
- 监控集群性能： 密切监控 Redis Cluster 的网络带宽、CPU 使用率等指标，及时发现和解决性能问题。
Pub/Sub 的应用场景
尽管有局限性，Pub/Sub 仍然在很多场景下非常有用：
- 实时消息推送： 例如，在聊天室、在线游戏等应用中，可以使用 Pub/Sub 将消息实时推送给客户端。
- 事件通知： 例如，当数据库发生变化时，可以使用 Pub/Sub 通知其他服务。
- 配置更新： 例如，当配置发生变化时，可以使用 Pub/Sub 将更新通知给所有应用实例。
总结
Redis Cluster 的 Pub/Sub 是一种简单而高效的消息广播机制。它适用于对消息可靠性要求不高，但对实时性要求较高的场景。在使用 Pub/Sub 时，我们需要注意其局限性，并采取相应的措施来缓解。
表格：Pub/Sub 与消息队列的对比
| 特性 | Pub/Sub | 消息队列 (例如 Redis Streams) |
|---|---|---|
| 消息持久化 | 无 | 有 |
| 消息确认 | 无 | 有 |
| 消息可靠性 | 低 | 高 |
| 消息消费模式 | 广播 | 点对点、发布/订阅、消费者组 |
| 应用场景 | 实时消息推送、事件通知、配置更新等 | 任务队列、事件驱动架构、微服务通信等 |
| 实现复杂度 | 简单 | 相对复杂 |
高级话题：模式订阅 (Pattern Subscription) 的坑
前面提到过，Redis Cluster 支持模式订阅，也就是使用通配符来订阅多个 Channel。例如，可以订阅 news.* 来接收所有以 news. 开头的 Channel 的消息。
但是，需要注意的是，Redis Cluster 的模式订阅性能可能较差。因为每个节点都需要维护一个模式订阅的列表，并且需要对每个消息进行模式匹配。在高并发场景下，这可能会导致 CPU 负载过高。
因此，建议尽量避免在生产环境中使用模式订阅。如果确实需要订阅多个 Channel，可以考虑使用客户端进行过滤，或者使用消息队列来实现更灵活的订阅策略。
代码示例：模式订阅
import redis
# 连接到 Redis Cluster (这里假设只有一个节点)
redis_cluster = redis.Redis(host='localhost', port=6379)
# 创建 PubSub 对象
pubsub = redis_cluster.pubsub()
def pattern_message_handler(message):
print(f"Received message on pattern {message['pattern'].decode()}: {message['data'].decode()}")
pubsub.psubscribe(**{'news.*': pattern_message_handler}) # 订阅所有以 'news.' 开头的频道，指定消息处理函数
thread = pubsub.run_in_thread(sleep_time=0.01) # 创建一个线程来处理消息
# 模拟 Publisher 发布消息
redis_cluster.publish('news.sports', 'Sports news!')
redis_cluster.publish('news.politics', 'Politics news!')
redis_cluster.publish('other.news', 'Other news!') # 不会被订阅到
time.sleep(1)
pubsub.punsubscribe('news.*') # 取消订阅所有以 'news.' 开头的频道
thread.stop()
print("Finished!")总结中的总结
Redis Cluster 的 Pub/Sub 功能就像一个高效但有点“粗糙”的广播系统。用得好，可以快速传递消息；用不好，可能会造成“交通堵塞”。理解其原理、掌握其用法、注意其局限性，才能真正发挥 Pub/Sub 的威力。
好了，今天的讲座就到这里。希望大家有所收获！记住，技术是为人类服务的，要用幽默的心态去学习和使用它。下次再见！