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缓存模式建议：
| 场景 | 建议 |
|---|---|
| 热点数据 | TTL + 随机抖动 |
| 空值 | 缓存 NULL（短 TTL） |
| 高并发 | 逻辑过期 + 异步重建 |
六、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通用建议
| 项 | 建议 |
|---|---|
| 连接池 | 单例 |
| 大 key | 严禁 |
| 热 key | 本地缓存 + Redis |
| Lua | 控制复杂度 |
| 序列化 | JSON / MsgPack |
十一、可以进一步扩展的功能
- Redis Stream 消费组
- RedLock 多实例锁（跨机房）
- Cache Aside / Write Through
- 二级缓存（Local + Redis）
- 监控（slowlog / latency）
