发布于 由admin

Redis 集群下的 Pub/Sub：跨槽位消息发布与订阅

各位观众，欢迎来到今天的Redis技术分享会！今天我们要聊的是一个挺有意思的话题：Redis集群下的Pub/Sub，特别是涉及到跨槽位消息发布与订阅的时候，这事儿就没那么简单了。

开场白：Pub/Sub，简单的快乐？

大家对Redis的Pub/Sub（发布/订阅）机制应该都不陌生吧？ 这玩意儿就好比广播电台，发布者（Publisher）负责发射信号，订阅者（Subscriber）负责接收信号。Redis的Pub/Sub在单机环境下用起来那是相当的丝滑，简单易懂，几行代码就能搞定。

# 单机版Pub/Sub 示例 (Python + redis-py)
import redis

# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)

# 发布者
def publisher():
    for i in range(5):
        message = f"Message #{i} from Publisher"
        r.publish("my_channel", message) # 发布消息到 "my_channel"
        print(f"Published: {message}")
        time.sleep(1)

# 订阅者
def subscriber():
    pubsub = r.pubsub()
    pubsub.subscribe("my_channel")

    for message in pubsub.listen():
        if message["type"] == "message":
            print(f"Received: {message['data'].decode('utf-8')}")

# 启动线程
import threading
import time

if __name__ == "__main__":
    t1 = threading.Thread(target=publisher)
    t2 = threading.Thread(target=subscriber)

    t1.start()
    t2.start()

    t1.join()
    t2.join()

这段代码简单粗暴，一个发布者往my_channel频道里扔消息，一个订阅者守着my_channel频道接收消息。Perfect！

集群：事情开始变得有趣…

但当我们把Redis升级到集群模式，事情就开始变得“有趣”起来了。Redis集群的核心思想是数据分片，也就是把数据分散存储在多个节点上，每个节点负责一部分的数据。这种方式提高了整体的吞吐量和可用性。

问题来了：Pub/Sub的频道信息存储在哪里？ 默认情况下，Redis集群的Pub/Sub频道信息是存储在发布者所在的节点上的。 这意味着，如果你的发布者和订阅者不在同一个节点上，而且他们的数据被分到了不同的槽位（Slot），那你就会遇到“跨槽位”的难题。

跨槽位的难题：广播电台信号覆盖不到？

想象一下，你开了一个广播电台（发布者），但是你的听众（订阅者）住在信号覆盖不到的区域，那你的广播就白播了。 这就是跨槽位Pub/Sub可能遇到的问题：订阅者可能无法收到发布者发出的消息，因为他们不在同一个节点上，或者说，他们监听的频道信息不在订阅者所在的节点上。

解决方案：让信号覆盖整个集群！

那么，如何解决这个问题呢？ 核心思想是：确保所有的订阅者都能收到消息，无论他们位于哪个节点。 这里有几种常见的解决方案：

  1. 客户端广播 (Client-Side Broadcasting)

    • 原理： 简单来说，就是发布者把消息发送给集群中的所有节点。每个节点再把消息转发给本地的订阅者。
    • 优点： 实现简单，不需要修改Redis服务器端的代码。
    • 缺点： 浪费带宽，增加了所有节点的负载。每个节点都会收到相同的消息，即使该节点上没有订阅者。
    • 适用场景： 小规模集群，消息量不大的场景。
    # 客户端广播示例 (Python + redis-py-cluster)
from rediscluster import RedisCluster
import time

# 集群配置
startup_nodes = [{"host": "127.0.0.1", "port": "7000"},
                {"host": "127.0.0.1", "port": "7001"},
                {"host": "127.0.0.1", "port": "7002"}]

# 连接Redis集群
rc = RedisCluster(startup_nodes=startup_nodes, decode_responses=True)

# 发布者
def publisher():
    for i in range(5):
        message = f"Message #{i} from Publisher (Client-Side Broadcasting)"
        # 获取所有master节点
        nodes = rc.get_nodes()
        for node in nodes:
            if node.is_master: # 只向master节点发布
                r = redis.Redis(host=node.host, port=node.port, decode_responses=True)
                r.publish("my_channel", message)
        print(f"Published (Client-Side Broadcasting): {message}")
        time.sleep(1)

# 订阅者 (每个节点都需要运行一个订阅者)
def subscriber():
    r = redis.Redis(host='localhost', port=7000, decode_responses=True) # 连接到本地节点
    pubsub = r.pubsub()
    pubsub.subscribe("my_channel")

    for message in pubsub.listen():
        if message["type"] == "message":
            print(f"Received (Client-Side Broadcasting): {message['data']}")

import threading

if __name__ == "__main__":
    t1 = threading.Thread(target=publisher)
    t2 = threading.Thread(target=subscriber) # 假设订阅者在7000端口的节点上

    t1.start()
    t2.start()

    t1.join()
    t2.join()

    在这个例子中，发布者会循环遍历集群中的所有Master节点，并向每个节点都发布消息。 订阅者只需要连接到本地节点并订阅频道即可。

  2. RESP3 的 Sharded Pub/Sub (Redis 7.0+)

    • 原理： Redis 7.0 引入了 RESP3 协议，并提供了一种原生的Sharded Pub/Sub机制。 通过 hash tag 强制让 channel 落在同一个 slot 上，从而实现集群环境下的pub/sub。
    • 优点： 性能好，是 Redis 官方推荐的解决方案。
    • 缺点： 需要升级到 Redis 7.0+，并且客户端需要支持 RESP3 协议。
    • 适用场景： Redis 7.0+ 环境，追求高性能的场景。
    # Sharded Pub/Sub 示例 (Python + redis-py, Redis 7.0+)
import redis

# 连接Redis (确保你的客户端支持 RESP3)
r = redis.Redis(host='localhost', port=6379, db=0, protocol=3) # 显式指定 protocol=3

# 发布者
def publisher():
    for i in range(5):
        message = f"Message #{i} from Publisher (Sharded)"
        r.publish("my_channel{hash_tag}", message) # 注意channel名称中的hash tag
        print(f"Published (Sharded): {message}")
        time.sleep(1)

# 订阅者
def subscriber():
    pubsub = r.pubsub()
    pubsub.subscribe("my_channel{hash_tag}") # 同样需要hash tag

    for message in pubsub.listen():
        if message["type"] == "message":
            print(f"Received (Sharded): {message['data'].decode('utf-8')}")

import threading
import time

if __name__ == "__main__":
    t1 = threading.Thread(target=publisher)
    t2 = threading.Thread(target=subscriber)

    t1.start()
    t2.start()

    t1.join()
    t2.join()

    关键点在于channel的名字 my_channel{hash_tag}， Redis会根据{}中的内容计算hash值，并把channel信息存储到对应的slot上。 只要保证所有channel的hash tag一致， 就能确保它们落在同一个slot上，从而实现集群环境下的pub/sub。

  3. 使用中间件 (Message Queue)

    • 原理： 引入一个专门的消息队列系统，例如 RabbitMQ、Kafka 等。发布者把消息发送到消息队列，订阅者从消息队列中消费消息。
    • 优点： 解耦发布者和订阅者，提供更强大的消息路由、持久化、可靠性等功能。
    • 缺点： 增加了系统的复杂性，需要维护额外的消息队列系统。
    • 适用场景： 需要高可靠性、复杂消息路由的场景。

    这个方案不属于 Redis 的范畴了，这里就不提供代码示例了。

  4. 定制化 Lua 脚本 (高级玩法)

    • 原理： 编写 Lua 脚本，在 Redis 服务器端执行发布和订阅操作。 Lua 脚本可以访问集群中的所有节点，从而实现跨槽位的消息传递。
    • 优点： 灵活性高，可以根据具体需求定制实现。
    • 缺点： 编写和维护 Lua 脚本需要一定的技术水平。
    • 适用场景： 需要高度定制化的 Pub/Sub 解决方案。

    Lua脚本示例：

    -- publish_to_all_nodes.lua
local channel = KEYS[1]
local message = ARGV[1]

local nodes = redis.call('CLUSTER', 'NODES')
for i, node_info in ipairs(nodes) do
    local node_id, node_address, flags = string.match(node_info, '(%S+) (%S+)%s+(.*)')
    if flags and string.find(flags, 'master') then
        local host, port = string.match(node_address, '(%S+):(%S+)')
        local redis_node = redis.connect(host, port)
        redis_node:call('PUBLISH', channel, message)
        redis_node:close()
    end
end
return 'OK'

    Python调用Lua脚本示例:

    import redis
import rediscluster

# 集群配置
startup_nodes = [{"host": "127.0.0.1", "port": "7000"},
                {"host": "127.0.0.1", "port": "7001"},
                {"host": "127.0.0.1", "port": "7002"}]

# 连接Redis集群
rc = rediscluster.RedisCluster(startup_nodes=startup_nodes, decode_responses=True)

# 加载Lua脚本
with open("publish_to_all_nodes.lua", "r") as f:
    publish_script = f.read()

publish_func = rc.register_script(publish_script)

# 发布者
def publisher():
    for i in range(5):
        message = f"Message #{i} from Publisher (Lua Script)"
        publish_func(keys=["my_channel"], args=[message])
        print(f"Published (Lua Script): {message}")
        time.sleep(1)

# 订阅者 (每个节点都需要运行一个订阅者)
def subscriber():
    r = redis.Redis(host='localhost', port=7000, decode_responses=True) # 连接到本地节点
    pubsub = r.pubsub()
    pubsub.subscribe("my_channel")

    for message in pubsub.listen():
        if message["type"] == "message":
            print(f"Received (Lua Script): {message['data']}")

import threading
import time

if __name__ == "__main__":
    t1 = threading.Thread(target=publisher)
    t2 = threading.Thread(target=subscriber) # 假设订阅者在7000端口的节点上

    t1.start()
    t2.start()

    t1.join()
    t2.join()

    这个例子中，Lua脚本会遍历集群中的所有master节点，并向每个节点都发布消息。 Python代码负责加载和执行Lua脚本。

总结：选择适合你的方案！

方案 优点 缺点 适用场景
客户端广播 实现简单 浪费带宽，增加节点负载 小规模集群，消息量不大
RESP3 Sharded Pub/Sub 性能好，Redis官方推荐 需要升级到Redis 7.0+，客户端需要支持RESP3 Redis 7.0+ 环境，追求高性能
消息队列中间件 解耦发布者和订阅者，提供更强大的消息路由、持久化、可靠性等功能 增加系统复杂性，需要维护额外的消息队列系统 需要高可靠性、复杂消息路由
定制化 Lua 脚本 灵活性高，可以根据具体需求定制实现 编写和维护 Lua 脚本需要一定的技术水平 需要高度定制化的 Pub/Sub 解决方案

选择哪种方案取决于你的具体需求和环境。 如果你只是想快速搞定，而且集群规模不大，消息量也不大，那么客户端广播可能是一个不错的选择。 如果你追求高性能，并且已经升级到了Redis 7.0+，那么Sharded Pub/Sub是最佳选择。 如果你需要更强大的消息路由和可靠性，那么消息队列中间件可能更适合你。 如果你需要高度定制化的解决方案，那么可以考虑使用Lua脚本。

最后的忠告：测试！测试！再测试！

无论你选择哪种方案，都一定要进行充分的测试！ 在生产环境上线之前，一定要模拟各种场景，确保你的Pub/Sub系统能够正常工作。 毕竟，谁也不想在关键时刻发现广播电台突然哑火了，对吧？

好了，今天的分享就到这里。 感谢大家的观看！ 希望今天的内容能够帮助大家更好地理解和应用Redis集群下的Pub/Sub。 如果大家有什么问题，欢迎随时提问。 我们下次再见！

发表回复

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