聊 Python 连接 Redis 这个话题我其实很早就想写了。这段时间帮朋友处理了几个线上缓存问题正好把踩过的坑、积累的经验串一遍——从最简单的redis-py安装到连接池、序列化、分布式锁再到用 Redis Desktop Manager 排查数据一整套流程今天全部讲透。这篇文章适合刚入门 Python 后端的同学照抄作业也适合已经写过几年代码但没系统梳理过 Redis 使用细节的朋友查漏补缺。只要你需要把数据放到 Redis 里、再从 Redis 里读出来并且希望这个过程中少踩几个坑那你就算来对地方了。1. 整体设计先搞清楚 Redis 在项目里到底扮演什么角色1.1 Redis 不是数据库备份它是快字当头的存储层很多人第一次接触 Redis 的时候心里会想这不就是个键值对数据库吗没错但如果你只是把它当数据库来用那就真的低估它了。Redis 的数据全部存在内存里所以读写速度可以做到几十万 QPS 级别比查 MySQL 快一个甚至两三个数量级。这也是为什么绝大数 Python 项目中Redis 不是替代业务数据库而是挡在数据库前面的一层加速缓冲层。我习惯把它理解成一个带过期时间的全局字典Python 的dict存进程内、重启就没Redis 存进程外、随时可共享、可持久化还能设置有效期。也就是说多个 Python 进程甚至多台机器上的 Python 服务都可以通过 Redis 共享同一份数据。这就是它解决的核心问题跨进程、跨机器的数据共享与高速读写。1.2 客户端库选型默认用 redis-py其他都是备选Python 连接 Redis 的库不少但你真正需要用到的绝大多数情况下就是官方推荐的redis-py。直接pip install redis就行这个库已经迭代十多年从 Redis 2.0 一路支持到 6.x/7.x线程安全、连接池内置、API 简单基本挑不出毛病。有人可能会问那hiredis呢hiredis是 C 语言写的 Redis 解析器作为redis-py的可选加速器存在。安装后通过redis.Redis(..., parser_classredis._HiredisParser)启用能把响应解析速度提升不少。实测下来在批量读写的压测场景里hiredis 对高吞吐有帮助但在普通 API 读写场景下体感差别不大。我的建议是装上了就用没装也不用特意折腾。1.3 连接参数五个参数决定你能否连上 Redis连接 Redis 的代码很简单但参数背后的细节值得先说清楚import redis r redis.Redis( host127.0.0.1, port6379, db0, passwordNone, decode_responsesTrue, socket_timeout5, socket_connect_timeout5, )首先是host和port。默认就是本机127.0.0.1:6379这个端口几乎是 Redis 行业通用端口生产环境为了安全往往会改掉同时也必配password。如果你在本机起 Redis 没设密码测试环境无所谓但只要是公司内部网络我建议不管多内网都加上requirepass因为 Redis 曾被攻击者利用未授权访问漏洞植入后门木马这个教训非常深刻。然后是db参数。Redis 默认有 16 个逻辑库编号 0 到 15。这里有一个我踩过的坑测试环境和生产环境的 Redis 配置如果不同代码里用了默认db0还好一旦配置了db1却读db0就会遇到明明存进去了怎么查不到的诡异问题。所以强烈建议项目配置文件里把 db 索引显式写出来不要靠默认值。decode_responses这个参数是最容易忽略的。默认情况下 Redis 返回给 Python 的是bytes字节串set的时候传字符串没问题但get回来就是bvalue这种带前缀b的字节串如果忘了解码把字节串直接拼进字符串里就会出现bhello这样的输出。很多新手在这卡半天其实加一行decode_responsesTrue让客户端自动把结果解码成字符串世界立刻清净了。socket_timeout和socket_connect_timeout是保命参数。如果 Redis 服务端卡住了没有超时设置Python 线程可能一直挂在等待响应上调用方跟着被拖死。设成 2 到 5 秒比较合理既能给慢查询留余地又能避免无限期阻塞。2. 核心细节连接、连接池与序列化方案选型2.1 最基础的连接写法从 set 和 get 开始连接建好之后最核心的存储操作其实就是两个方法set和get。直接看最朴素的用法import redis r redis.Redis(host127.0.0.1, port6379, db0, decode_responsesTrue) r.set(user:10001:name, 张三) r.set(user:10001:age, 28) name r.get(user:10001:name) age r.get(user:10001:age) print(name, age)这里有一点要提醒Redis 的set只能存字符串、整数、浮点数这类简单值它不会自动帮你把 Python 的字典、列表转成字符串。如果你直接r.set(key, {a: 1})redis-py会抛异常——因为它收到的是 dict 类型没法直接转成字节串。那怎么存复杂结构两种思路一种是存之前自己json.dumps()读出来再json.loads()另一种更省心用后面会说的序列化封装。不过在讲序列化之前先把连接池这件事说透因为它直接影响你后面所有操作的稳定性。2.2 连接池能不能扛住高并发就看这一步了我见过不少项目代码里每个函数都这么写def get_user_name(user_id): r redis.Redis(host127.0.0.1, port6379, db0) return r.get(fuser:{user_id}:name)每次调用都创建新的连接用完也不关。这在低并发下看起来没事但一旦并发量上来TCP 连接的建立和销毁会占用大量资源Redis 端也会因为连接数暴增出现max number of clients reached的报错。我的建议是全局共享一个连接池连接池内部复用连接。import redis pool redis.ConnectionPool( host127.0.0.1, port6379, db0, decode_responsesTrue, max_connections50, ) r redis.Redis(connection_poolpool)ConnectionPool会预先创建一批连接放在池子里每次用的时候借一条用完归还。max_connections50这个值可以根据业务 QPS 来调不用拍脑袋。经验公式并发峰值请求数除以单个请求的 Redis 往返次数再乘以 1.5 的余量系数。假设峰值 QPS 是 1000每次请求只操作 1 次 Redis那同时需要的连接数大概就是 1000 个连接同时被占用——不对这里要算的是并发瞬时连接数不是 QPS。更简单的估算方式如果服务有 100 个线程每个线程同时只会占用一条 Redis 连接max_connections100就足够了。没必要设太大连接池本身也会占用 Redis 端内存资源。2.3 序列化存 JSON、pickle 还是 msgpack前面说过Redis 只认字符串和字节串。要把 Python 的字典、列表、对象存进去序列化这一步躲不掉。我梳理了三套方案各有各的适用场景。方案一JSON 序列化。最通用跨语言兼容性最好。存的时候json.dumps(obj)读出来json.loads(obj)。注意 Python 的json.dumps默认会把中文变成\u转义序列虽然不影响存储和读取但你在 Redis Desktop Manager 里看数据会觉得特别别扭。解决办法是json.dumps(obj, ensure_asciiFalse)。方案二Pickle 序列化。Python 专属能把任意对象包括自定义类实例直接序列化非常方便。但缺点很致命跨语言不可读而且存在安全风险——千万不能用pickle.loads()去加载不可信来源的数据这可能触发任意代码执行。所以 pickl 我只用在内部缓存、可信数据源场景。方案三msgpack。一个比 JSON 更紧凑的二进制序列化格式处理速度和体积都优于 JSON。redis-py还自带了一个redis.msgpack封装不过现实中我更推荐手动处理因为看懂底层细节才好在问题排查时不慌。我比较推荐的做法是写一个小工具类把序列化逻辑收敛到一处import json import redis class RedisCache: def __init__(self, client): self.client client def set_data(self, key, value, expireNone): payload json.dumps(value, ensure_asciiFalse) if expire: self.client.set(key, payload, exexpire) else: self.client.set(key, payload) def get_data(self, key): payload self.client.get(key) if payload is None: return None return json.loads(payload)这样业务代码里直接cache.set_data(user:info:10001, {name: 张三, age: 28}, expire300)完全感知不到序列化的细节。2.4 Pipeline 和事务批量写入的正确姿势存一条数据调用一次set没问题但如果有几十条甚至几百条数据要写逐条执行就太低效了。每次操作都是一次网络往返100 条数据就是 100 次网络耗时。这时候用 Pipeline 把命令打包发给 Redis一次网络往返搞定所有命令pipe r.pipeline(transactionTrue) for i in range(1000): pipe.set(fbatch:key:{i}, fvalue:{i}) pipe.execute()transactionTrue的意思是打包的命令在一个事务里执行要么全部成功要么全部失败中间不会被别的客户端命令插入。这在做批量初始化、多键更新时要特别注意能避免写了一半另外一半没写进去带来的脏数据问题。不过pipeline也不是银弹。如果命令数量特别大比如一次打包几万条Redis 端处理这些命令期间会阻塞其他命令的执行。我踩过一次坑凌晨跑批量任务打包了 5 万条set执行的时候线上接口延迟飙到几秒。后来改成每 1000 条一次分批执行问题立刻消失。所以 Pipeline 的使用原则是单批命令控制在 500 到 1000 条别贪多。3. 实操过程五种数据结构的存储与读取Redis 不止有 String 类型。很多面试题会问Redis 有哪些数据类型但真实项目里用得最多的其实是五种String、Hash、List、Set、ZSet。下面逐个讲存储时的代码写法和使用场景这部分每个 Python 后端都应该熟练掌握。3.1 String最常见的缓存写入与读取String 类型不仅能存简单的键值还能做原子自增操作。比如记录用户访问次数r.incr(user:10001:visit_count) r.incrby(user:10001:visit_count, 5)incr是原子操作多个进程同时执行也不会计数丢失。这个特性非常适合做计数器、限流统计、库存扣减。setnx系列命令也特别常用它只在键不存在时才设置值后面讲分布式锁时会用到。存字符串时要注意set方法的参数名。redis-py里设置过期时间的参数是ex单位是秒还有一个px参数单位是毫秒。如果概念搞混了你可能本来想设置 10 秒过期结果写成了px10实际 10 毫秒后键就消失了。这个问题我在公司里抓到过不止一次r.set(temp:key, value, ex60) # 60 秒过期 r.set(temp:key, value, px60000) # 60000 毫秒等价于 60 秒3.2 Hash对象结构用哈希类型最自然如果你要存的是一个用户对象包含姓名、年龄、积分等多个字段用 String 存整个 JSON 可以但每次改一个字段都要读出整个对象、反序列化、修改、再写回去特别浪费。Hash 类型就是专门解决这个问题的r.hset(user:10001, mapping{ name: 张三, age: 28, score: 100 })读取支持只取某个字段age r.hget(user:10001, age) # 返回 28 name r.hget(user:10001, name)如果是热更新比如用户积分从 100 变成 120用hincrby一步到位r.hincrby(user:10001, score, 20)这里有一个非常容易踩的坑同一个对象的业务字段有些用 String 存、有些用 Hash 存键还长得一样。比如一段代码hgetall(user:10001)另一段代码get(user:10001)两个命令作用在同一键上会互相覆盖导致数据变成串味。解决办法是命名规范上做区分比如 Hash 键统一叫user:10001:hashString 键叫user:10001:name不要混用同一个键名。3.3 List先入先出的消息管道List 类型底层是链表支持左右两端插入和弹出。最经典的使用场景是轻量级队列。比如一个爬虫项目任务列表用 List 存储r.lpush(task:queue, url:1) r.lpush(task:queue, url:2) task r.rpop(task:queue) # 取出 url:1生产者用lpush往左边塞消费者用rpop从右边取正好先进先出。如果要阻塞等待可以换brpop带超时时间避免空轮询消耗 CPU 和网络task r.brpop(task:queue, timeout5)brpop的返回值是一个元组(key, value)不是单纯的值初用者经常在这栽跟头。3.4 Set 和 ZSet去重、标签与排行榜Set 是无序集合自动去重适合做标签系统、去重名单r.sadd(user:10001:tags, vip, active, vip) # 重复的 vip 自动忽略 r.smembers(user:10001:tags) # 返回两个元素ZSet 是有序集合每个成员带一个分值。最典型的场景是排行榜r.zadd(game:rank:2024, {player_a: 1000, player_b: 900}) r.zincrby(game:rank:2024, 100, player_a)读取前三名top3 r.zrevrange(game:rank:2024, 0, 2, withscoresTrue)zrevrange按分数从高到低取withscoresTrue让结果带上分数。这里需要注意返回的分数是字符串转浮点数的时候要做一次float()转换我在线上日志里看到过很多次TypeError就是因为忘了这一步。为了帮助快速选型我把五种数据类型整理成了一个对比表数据类型底层结构典型场景常用命令注意事项String字节数组缓存、计数器、限流set/get/incr/setnx最通用适合单值Hash哈希表对象、用户信息hset/hget/hgetall/hincrby字段级更新省带宽List双向链表消息队列、最近列表lpush/rpop/brpop注意阻塞命令返回值Set哈希集合标签、去重sadd/smembers/sismember无序注意成员量大时的内存ZSet跳表哈希排行榜、优先级zadd/zrange/zrevrange分值注意类型转换3.5 过期时间与键管理不是所有数据都应该永久存活Redis 存的是内存数据内存再大也有限。一条缓存永远不过期最终的结果就是 OOM 崩溃。这里分享几条我在项目中固化的经验需要缓存的数据能设置过期时间的必须设置过期时间。比如用户信息缓存 10 分钟商品详情缓存 1 小时根据业务实时性要求来定。批量设置过期时间时加一个随机偏移量。比如基础过期时间是 10 分钟实际设置时在基础值上加random.randint(0, 60)秒。这样做的目的是避免大规模缓存同时失效——同时失效就相当于同时打穿到数据库俗称缓存雪崩。删除键用delete方法后面讲缓存治理时还会提到。还有一个很隐蔽的坑redis-py的set(name, value, ex60)是从设置时开始计时的 60 秒。每次重新set都会刷新过期时间。如果你希望键在活跃后自动续期可以每次访问后重新设置过期时间但要注意这不是线程安全的极端并发下仍有失效可能。4. 进阶场景缓存治理、分布式锁和连接模式升级4.1 缓存穿透、击穿、雪崩三个让后端崩溃的经典问题这三个名词在 Redis 面试题里出现频率极高真实项目中却真的能遇到。我把它们放一起讲因为理解了这三个问题你才知道为什么 Redis 存储数据这件事并不仅仅是存进去、取出来这么简单。缓存穿透大量请求查询一个不存在的键缓存始终未命中请求全部落到数据库。解决思路是布隆过滤器拦截或者在缓存中为不存在的数据也存一个空值占位比如set(user:99999, null, ex60)。缓存击穿某个热点键过期的瞬间大量请求同时发现缓存失效一起打到数据库。解决思路是互斥锁或逻辑过期。用 Python 实现互斥锁比较简单def get_user_data(user_id): key fuser:data:{user_id} data r.get(key) if data is not None: return data lock_key flock:{key} if r.set(lock_key, 1, nxTrue, ex10): try: data query_from_mysql(user_id) r.set(key, data, ex300) return data finally: r.delete(lock_key) else: time.sleep(0.1) return get_user_data(user_id)缓存雪崩大量键在同一时间过期或者 Redis 实例宕机导致请求直接冲击数据库。处理方案是过期时间加随机偏移量同时做多级缓存兜底。这三个问题不是危言耸听。我在公司处理过一次线上事故双十一活动开始的瞬间一个热门商品的详情缓存恰好过期结果后端数据库连接数瞬间打满接口大面积超时。后来就是靠互斥锁 过期时间随机化才彻底解决的。4.2 分布式锁用 Redis 实现多进程互斥Python 的多线程可以用threading.Lock但多台机器上的多个 Python 进程怎么互斥方案之一就是 Redis 分布式锁。原理是利用setnx命令的原子性谁先设置成功谁就拿到锁。def acquire_lock(lock_key, timeout30): # nxTrue 表示只有当键不存在时才设置成功 # pxtimeout 表示锁的自动过期时间防止持有锁的进程崩溃后死锁 return r.set(lock_key, locked, nxTrue, pxtimeout) def release_lock(lock_key): r.delete(lock_key)注意这里有个关键细节set的nxTrue和pxtimeout必须同时写在一个命令里这样才能保证原子性。如果先setnx再expire两条命令分开执行进程在中间崩掉会导致锁永不过期所有拿到锁的请求全部卡死。释放锁的时候要注意只有锁的持有者才能释放。如果进程 A 执行业务超时锁自动过期了进程 B 后来拿到锁这时候进程 A 才执行delete就会把进程 B 的锁误删。规避方式是在设置锁时放一个唯一标识释放时先校验标识再删除。这个逻辑用 Lua 脚本保证原子性最稳妥。Python 代码里redis-py的eval方法可以执行 Lua 脚本这里就不展开写了但思路一定要有。4.3 从单机到主从、集群连接方式怎么变我最初入门时只在本地起一个 Redis 单实例后来做高可用项目发现生产环境至少是主从架构。主从复制解决了单点故障问题主节点写从节点读主节点挂了从节点顶上。Python 连接主从集群时redis-py提供了Redis和RedisCluster两种客户端。主从模式下通常的配置是写操作走主节点读操作走从节点master redis.Redis(hostredis-master, port6379, decode_responsesTrue) slave redis.Redis(hostredis-slave, port6379, decode_responsesTrue) master.set(user:1, data) # 写主库 value slave.get(user:1) # 读从库但这个方案有一个一致性问题主从同步有延迟刚写入的数据可能在从库上暂时读不到。对一致性要求高的场景写后立刻读要走主库。这是我实际项目里调了很久才发现的坑写完后从库读返回空值排查了半天才意识到是主从延迟。如果你用云厂商提供的托管 Redis 服务通常在控制台直接给出连接地址Python 代码里的host填这个地址就行底层的主从切换由云平台处理。自己搭建 Docker 环境时网上有大量docker compose起 Redis 主从的方案可以参考核心是配置主从复制关系replicaof redis-master 63794.4 监控好帮手Redis Desktop Manager 怎么用写代码存数据只是第一步你总得亲眼看到数据在 Redis 里长什么样。这里我强烈推荐 Redis Desktop Manager这个工具可以直观地浏览所有键支持按模式搜索还能查看键的过期时间和内存占用。查排除问题的时候非常方便。简单介绍一下用法下载安装后新建连接填host、port、password连接成功左边会显示所有 db 编号点进去就能看到一堆键。搜索某个键时直接在过滤框填模式比如user:10001:hash立刻就能筛选出来。很多诡异问题比如我明明 set 了为什么查不到用 RDM 一查往往就会发现是 db 选错了或者键名拼错了。RDM 还能执行命令行相当于可视化的redis-cli调试脚本的时候直接在里面跑get、ttl这些命令十分顺手。5. 常见问题与排查技巧实录这一节的内容全部来自我实际工作中踩坑之后留下的排查笔记。每个问题都有对应思路建议遇到类似情况时按表格对照检查。5.1 连不上ConnectionRefusedError 与超时最常见的问题就是连接被拒绝。排查顺序固定这几步确认 Redis 服务是否启动终端执行redis-cli ping返回PONG说明服务正常。确认端口是否被占用或配置错误lsof -i :6379或 Windows 下的netstat -ano | findstr 6379。确认密码是否正确如果配置文件设置了requirepass连接时必须传password否则会报NOAUTH Authentication required。确认防火墙是否放行端口云服务器上跑 Redis安全组规则必须放行对应端口。还有一个隐蔽问题Redis 默认只绑定127.0.0.1如果你把连接地址写成了远程 IP 或公网 IP即使防火墙放行了也连不上。要远程连接必须在 Redis 配置里把bind改成实际监听地址像这样bind 0.0.0.0但这里我必须强烈提醒把 Redis 暴露到公网且不设密码等于裸奔这不是危言耸听。我在测试服务器上也遇到过扫描器尝试爆破的情况Monti 木马就是利用 Redis 未授权访问入侵服务器的典型案例。所以生产环境务必加密码 限制来源 IP 改默认端口三件套缺一不可。5.2 数据读出乱码bytes 和解码的关系你有没有遇到过get出来的值是bhello这样的原因就是我们前面说的decode_responses没有设置。还有一个高频场景存了中文读取时显示正常但 JSON 序列化后变成了\u5f20\u4e09。这种转义写法不影响数据本身如果你实在看着难受用ensure_asciiFalse重新序列化即可。5.3 键过期了为什么内存还很高Redis 删除过期键有两种机制惰性删除和定期删除。惰性删除是访问到键时才删定期删除是每 10 秒抽样删除部分过期键。如果一个键很大、量又多过期后内存不会立刻全部释放。执行redis-cli info memory可以看到内存碎片率和键数量必要时手动执行memory purge。另外如果一个键长期不访问却又一直存在可以设置内存淘汰策略来兜底maxmemory-policy allkeys-lru这个配置表示内存满时按 LRU 算法淘汰不常用的键保证服务不会因为内存写满而崩溃。5.4 命令越来越慢慢查询和 bigkey 排查如果某个get、hgetall突然变得很慢先别急着骂网络。执行redis-cli slowlog get 10查看最近 10 条慢查询命令看看是不是某个 100 万字段的 Hash 被整个hgetall拉了出来。大 key 是 Redis 性能杀手一个哈希里有几十万个字段每次读取都耗时巨大还阻塞其他命令。排查大 key 的方式很直接redis-cli --bigkeys这个命令会扫描整个实例并汇报最大的 key。业务侧的处理思路是大 Hash 拆成小 Hash按固定字段数分段或者把只读冷数据从 Redis 移到 CDN 或本地缓存。5.5 面试官常问缓存一致性怎么保证这个话题在 Redis 面试题里必考但实际项目中很多人没想清楚。一条数据既在 MySQL 里、又在 Redis 里修改 MySQL 后 Redis 怎么办三种常规思路先删 Redis 缓存再更新 MySQL。删除成功后下次查询自然会重新回填这是比较保险的做法。先更新 MySQL再删除 Redis 缓存。这种方式可能导致短时间内缓存里是旧数据需要配合延迟双删。给缓存加短过期时间。即使不一致最多持续几秒到几十秒。我自己项目里的经验是对一致性要求不高的读多场景更新数据库后主动删除缓存键让下次读取自然回源对一致性要求极高的场景直接不走 Redis全量查数据库。兼顾性能和数据一致性本身就是个权衡题没有银弹。5.6 序列化版本兼容升级后读不出旧数据这个坑比较罕见但我确实遇到过项目从 msgpack 0.5 升级到 0.6 后缓存里的旧数据反序列化失败。原因是新版本对某些二进制格式的解释变了。排查思路是拿到 Redis 里的原始字节串pickle.loads或msgpack.unpackb逐层试定位到不兼容的具体格式。经验是高频变更的缓存数据结构最好在 key 中带版本号比如user:info:v1:10001升级时通过版本号强制全量刷新。这是成本最低、最不容易出错的方案。写在最后Python 连接 Redis 这件事如果只看最简单的set、get半小时就能学会。但真实项目里连接池怎么配、序列化选什么、缓存怎么治理、分布式锁怎么写、线上慢查询怎么排查每一条都是实打实踩出来的经验。我个人的建议是从最小的 Demo 开始先在本机装一个 Redis用redis-py把五种数据类型各写一遍再用 Redis Desktop Manager 观察数据变化最后再把连接池、序列化封装到你的项目里。这套流程走下来Redis 在你手里就不是一个会用的工具而是一个你用得很稳的中间件。后续如果你进一步接触 Redis 7.0 的 RedisJSON、RedisSearch 模块或者用docker compose搭主从复制环境你会发现万变不离其宗——连接、存储、排查始终是三个最核心的主题。