Python 整合 Redis 哨兵（Sentinel）与集群（Cluster）实战指南

Python 整合 Redis 哨兵（Sentinel）与集群（Cluster）实战指南

面向生产环境的 Redis Python 工程化示例，涵盖：

• 哨兵模式 & 集群模式自动切换
• 常用数据结构（String / Hash / List / Set / ZSet）
• 分布式锁（可重入 / 自动续期 / Lua）
• 发布订阅（Pub/Sub）
• 原子计数 / 限流 / 缓存模式
• 健康检查、连接池、异常处理、最佳实践

一、技术选型说明

1. Python Redis 客户端

redis-py >= 5.x

• 官方维护
• 原生支持Sentinel
• 原生支持Redis Cluster
• 支持 pipeline / lua / pubsub

安装：

pipinstallredis>=5.0.0

二、项目结构（推荐生产级）

redis_client/ ├── __init__.py ├── config/ │ ├── __init__.py │ └── settings.py # Redis 配置（哨兵 / 集群） │ ├── core/ │ ├── __init__.py │ ├── client.py # Redis 客户端工厂（核心） │ ├── connection.py # 连接池管理 │ └── exceptions.py # 自定义异常 │ ├── features/ │ ├── __init__.py │ ├── cache.py # 通用缓存封装 │ ├── lock.py # 分布式锁 │ ├── counter.py # 计数器 / 限流 │ ├── pubsub.py # 发布订阅 │ └── structures.py # Redis 各数据结构封装 │ ├── utils/ │ ├── __init__.py │ ├── serializer.py # JSON / MsgPack │ └── time.py │ └── examples/ ├── sentinel_demo.py ├── cluster_demo.py └── pubsub_demo.py

三、Redis 配置（settings.py）

# redis_client/config/settings.pyREDIS_MODE="sentinel"# sentinel | cluster | standalone# Sentinel 配置REDIS_SENTINEL={"sentinels":[("10.0.0.1",26379),("10.0.0.2",26379),],"service_name":"mymaster","db":0,"password":None,"socket_timeout":3,}# Cluster 配置REDIS_CLUSTER={"startup_nodes":[{"host":"10.0.0.10","port":6379},{"host":"10.0.0.11","port":6379},],"password":None,"socket_timeout":3,}

四、Redis 客户端工厂（核心）

# redis_client/core/client.pyfromredisimportRedisfromredis.sentinelimportSentinelfromredis.clusterimportRedisClusterfromredis_client.config.settingsimport(REDIS_MODE,REDIS_SENTINEL,REDIS_CLUSTER,)classRedisClientFactory:_client=None@classmethoddefget_client(cls):ifcls._client:returncls._clientifREDIS_MODE=="sentinel":sentinel=Sentinel(REDIS_SENTINEL["sentinels"],socket_timeout=REDIS_SENTINEL["socket_timeout"],password=REDIS_SENTINEL["password"],)cls._client=sentinel.master_for(service_name=REDIS_SENTINEL["service_name"],db=REDIS_SENTINEL["db"],decode_responses=True,)elifREDIS_MODE=="cluster":cls._client=RedisCluster(startup_nodes=REDIS_CLUSTER["startup_nodes"],password=REDIS_CLUSTER["password"],decode_responses=True,)else:cls._client=Redis(host="localhost",port=6379)returncls._client

优点：

• 调用方无感知哨兵 / 集群
• 自动主从切换
• 单例连接池

五、通用缓存封装（cache.py）

# redis_client/features/cache.pyimportjsonfromredis_client.core.clientimportRedisClientFactoryclassCache:def__init__(self):self.redis=RedisClientFactory.get_client()defset(self,key,value,ttl=None):val=json.dumps(value)self.redis.set(key,val,ex=ttl)defget(self,key,default=None):val=self.redis.get(key)returnjson.loads(val)ifvalelsedefaultdefdelete(self,key):self.redis.delete(key)defexists(self,key)->bool:returnself.redis.exists(key)==1

缓存模式建议：

六、Redis 数据结构封装（structures.py）

# redis_client/features/structures.pyfromredis_client.core.clientimportRedisClientFactoryclassRedisStructures:def__init__(self):self.redis=RedisClientFactory.get_client()# Stringdefincr(self,key,amount=1):returnself.redis.incr(key,amount)# Hashdefhset(self,name,key,value):self.redis.hset(name,key,value)defhget(self,name,key):returnself.redis.hget(name,key)# Listdeflpush(self,key,value):self.redis.lpush(key,value)defrpop(self,key):returnself.redis.rpop(key)# Setdefsadd(self,key,value):self.redis.sadd(key,value)# ZSetdefzadd(self,key,mapping:dict):self.redis.zadd(key,mapping)

七、分布式锁（lock.py）

1. Lua 保证原子性

# redis_client/features/lock.pyimportuuidimporttimefromredis_client.core.clientimportRedisClientFactory UNLOCK_SCRIPT=""" if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end """classRedisLock:def__init__(self,key,ttl=10):self.redis=RedisClientFactory.get_client()self.key=f"lock:{key}"self.ttl=ttl self.value=str(uuid.uuid4())defacquire(self,retry=3):for_inrange(retry):ifself.redis.set(self.key,self.value,nx=True,ex=self.ttl):returnTruetime.sleep(0.1)returnFalsedefrelease(self):self.redis.eval(UNLOCK_SCRIPT,1,self.key,self.value)

特点：

• 防误删
• 支持哨兵 / 集群
• Lua 原子性

八、发布订阅（pubsub.py）

# redis_client/features/pubsub.pyimportthreadingfromredis_client.core.clientimportRedisClientFactoryclassRedisPubSub:def__init__(self):self.redis=RedisClientFactory.get_client()defpublish(self,channel,message):self.redis.publish(channel,message)defsubscribe(self,channel,handler):pubsub=self.redis.pubsub()pubsub.subscribe(channel)deflisten():formsginpubsub.listen():ifmsg['type']=='message':handler(msg['data'])threading.Thread(target=listen,daemon=True).start()

⚠️ 注意：

• Pub/Sub不可靠（断线即丢）
• 生产消息推荐Redis Stream / MQ

九、计数器 & 限流（counter.py）

# redis_client/features/counter.pyimporttimefromredis_client.core.clientimportRedisClientFactoryclassRateLimiter:def__init__(self,key,limit,window):self.redis=RedisClientFactory.get_client()self.key=key self.limit=limit self.window=windowdefallow(self):now=int(time.time())pipe=self.redis.pipeline()pipe.incr(self.key)pipe.expire(self.key,self.window)count,_=pipe.execute()returncount<=self.limit

十、生产级最佳实践总结

Redis 哨兵

• 客户端必须通过 Sentinel 发现主节点
• 禁止直连 master IP

Redis Cluster

• Key 必须使用hash tag保证多 key 原子：

user:{123}:name user:{123}:age

通用建议

十一、可以进一步扩展的功能

• Redis Stream 消费组
• RedLock 多实例锁（跨机房）
• Cache Aside / Write Through
• 二级缓存（Local + Redis）
• 监控（slowlog / latency）