Redis-哨兵（Sentinel）和集群（Cluster）链接方式

运维小路

发布于 2026-01-28 14:32:23

1930

文章被收录于专栏：运维小路运维小路

作者介绍：简历上没有一个精通的运维工程师。请点击上方的蓝色《运维小路》关注我，下面的思维导图也是预计更新的内容和当前进度(不定时更新)。

数据库是一个系统（应用）最重要的资产之一，所以我们的数据库将从以下几个数据库来进行介绍。

MySQL

PostgreSQL

MongoDB

Redis（本章节）

Etcd

我们前面介绍了单机，主从，哨兵，集群多种Redis部署方式，他们的链接方式也是不一样的，所以我们用一个小节来介绍程序代码应该如何链接不同方式的Redis，后续都时使用Python代码，这里为了简单所以配置都在代码里面。

基础准备

# 安装pip
yum -y install python-pip
# 安装依赖，这个版本才支持低版本的redis
pip install redis==3.5.3
# 如果需要连接集群
pip install redis-py-cluster

单机Redis链接

# coding: utf-8
import redis

# 基础连接
redis_client = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    password=None,  # 如果有密码
    decode_responses=True  # 自动解码返回字符串
)

# 测试连接
try:
    redis_client.ping()
    print("单机Redis连接成功")
except Exception as e:
    print("连接失败:", str(e))

# 基本操作示例
redis_client.set('key', 'value')
value = redis_client.get('key')
print(value)

主从Redis链接

如果你不需要读写分立，那么可以和单机Redis链接方式一样，

import redis

master = redis.Redis(host='192.168.31.196', port=6379)
slave = redis.Redis(host='192.168.31.197', port=6379)

print("Redis Master-Slave Test")
print("Master: 192.168.31.196:6379")
print("Slave: 192.168.31.197:6379")
print("-" * 30)

# Write to master
print("1. Write to master...")
master.set('test_key', 'test_value')
print("Write success: key=test_key, value=test_value")

# Read from slave
print("\n2. Read from slave...")
result = slave.get('test_key')
if result:
    print("Read success: %s" % result)
else:
    print("Read failed")

print("\n3. Test complete")

哨兵

这里的Redis代码直接链接的是Redis哨兵地址，并没有直接链接Redis实例，当主从切换以后，他还是可以获取当前的Master，进行写入操作。

# -*- coding: utf-8 -*-
import redis
from redis.sentinel import Sentinel
import time

# 哨兵配置
sentinel = Sentinel([
    ('192.168.31.196', 26379),
    ('192.168.31.197', 26379)，
    ('192.168.31.197', 26379)
])
print("Redis哨兵写入测试")
print("=" * 40)
try:
    # 第一次写入
    print("第一次写入...")
    master = sentinel.master_for('mymaster', decode_responses=True)
    master_info = sentinel.discover_master('mymaster')
    print("当前主节点: %s:%s" % (master_info[0], master_info[1]))

    master.set('test_key', 'value1')
    print("写入: test_key=value1")

    # 等待30秒
    print("\n等待30秒...")
    for i in range(30):
        time.sleep(1)
        if (i+1) % 10 == 0:
            print("已等待 %d 秒" % (i+1))

    # 第二次写入
    print("\n第二次写入...")
    master = sentinel.master_for('mymaster', decode_responses=True)
    new_master_info = sentinel.discover_master('mymaster')
    print("当前主节点: %s:%s" % (new_master_info[0], new_master_info[1]))

    # 检查是否切换
    if master_info != new_master_info:
        print("主节点已切换!")
        print("原主节点: %s:%s" % (master_info[0], master_info[1]))
        print("新主节点: %s:%s" % (new_master_info[0], new_master_info[1]))
    else:
        print("主节点未切换")

    master.set('test_key', 'value2')
    print("写入: test_key=value2")

    # 显示最终数据
    value = master.get('test_key')
    print("最终数据: test_key=%s" % value)

    print("\n测试完成!")

except Exception as e:
    print("\n错误: %s" % str(e))
    print("请检查哨兵配置和连接")

集群

当然这里只写了一个地址，真实环境肯定会写满地址。

# -*- coding: utf-8 -*-
from rediscluster import RedisCluster
import redis

# 连接到集群
rc = RedisCluster(
    startup_nodes=[{"host": "127.0.0.1", "port": 7001}],
    decode_responses=True,
    skip_full_coverage_check=True
)

print("Redis集群槽位分布测试")
print("=" * 50)

# 测试几个键
test_data = {
    "user:1": "Alice",
    "user:100": "Bob",
    "user:1000": "Charlie",
    "order:1": "Order1",
    "order:100": "Order100",
    "item:A": "ItemA",
    "item:Z": "ItemZ"
}

print("\n键的槽位分布:")
print("Key\t\t\tSlot\tNode")
print("-" * 40)

for key, value in test_data.items():
    rc.set(key, value)

    # 获取槽位
    r = redis.Redis(host='127.0.0.1', port=7001, decode_responses=True)
    slot = r.execute_command('CLUSTER KEYSLOT', key)

    # 判断节点
    if 0 <= slot <= 5460:
        node = "7001"
    elif 5461 <= slot <= 10922:
        node = "7002"
    else:
        node = "7003"

    print("%-15s\t%5d\t%s" % (key, slot, node))

print("\n" + "=" * 50)
print("槽位范围:")
print("7001: 0 - 5460")
print("7002: 5461 - 10922")
print("7003: 10923 - 16383")