Redis Sentinel故障转移后如何实时更新Redis主节点对象

你当前的实现仅在程序启动阶段调用一次discover_master()获取主节点地址，后续所有操作都绑定到这个固定地址的Redis实例。Sentinel触发故障转移后，旧主节点会被降级为从节点甚至下线，硬编码的固定连接自然无法自动适配新的主节点。

redis-py 库的Sentinel类本身内置了主节点自动感知能力，不需要手动维护主节点地址、监听切换事件，直接使用官方提供的连接代理即可：

• 不要手动调用discover_master()后创建独立的固定Redis实例
• 直接调用Sentinel实例的master_for()方法获取主节点操作对象，该对象会在执行命令前自动校验当前主节点地址，故障转移完成后会自动将连接切换到新晋升的主节点，对业务代码完全透明。

对应实现代码：

from redis.sentinel import Sentinel

# 初始化Sentinel集群连接
sentinel = Sentinel(
    [('192.168.10.1', 26379), ('192.168.10.2', 26379), ('192.168.10.3', 26379)],
    socket_timeout=0.1,
    retry_on_timeout=True,
    db=0
)

# 获取主节点代理对象，全程不需要手动解析host/port创建Redis实例
master = sentinel.master_for("master-name", socket_timeout=0.1)

# 后续所有Redis操作直接调用该对象即可，故障转移后自动切换新主节点
master.get("data")

如果业务需要在主节点切换时触发额外操作（比如日志告警、本地缓存清理），可以启动独立守护线程订阅Sentinel的事件频道，感知切换事件：

import threading

def listen_switch_event(sentinel, target_master_name):
    pubsub = sentinel.pubsub()
    # 订阅Sentinel主节点切换事件频道
    pubsub.subscribe("+switch-master")
    for event in pubsub.listen():
        if event.get("type") != "message":
            continue
        msg = event["data"].decode()
        # 消息格式：<主节点名> <旧IP> <旧端口> <新IP> <新端口>
        parts = msg.split()
        if len(parts) < 5 or parts[0] != target_master_name:
            continue
        old_host, old_port, new_host, new_port = parts[1:]
        # 在此处添加自定义业务逻辑
        print(f"主节点已完成切换：{old_host}:{old_port} -> {new_host}:{new_port}")

# 启动守护线程运行监听逻辑，不会阻塞主业务流程
threading.Thread(
    target=listen_switch_event,
    args=(sentinel, "master-name"),
    daemon=True
).start()

注意：99%的业务场景下直接使用master_for()即可满足高可用需求，不需要自行实现事件监听、主动刷新主节点地址的逻辑。自行维护主节点对象容易出现连接泄漏、并发请求打到旧主节点、重试逻辑异常等问题。

内容的提问来源于stack exchange，提问作者rhoitjadhav