做Python后端开发的人迟早都要跟Redis打交道。它速度快、数据结构丰富、部署也简单很多项目里缓存、session、计数、队列、去重这些活儿都是用Redis扛着的。这篇东西我就基于实际经验把Python连接Redis数据库进行增删改查完整过一遍把常用的方法、参数背后的原因、容易踩的坑全交代清楚。无论你是刚学Python还是写了好几年业务代码都能在这里找到可以直接照着用的内容。先说一个最基本的判断如果你只是存个变量、做个数组那用内存就够了Redis没必要上。但一旦你需要多进程共享状态、多个服务同步数据、或者把热点数据挡住数据库压力Redis就是最顺手的那个工具。它本质上是一个跑在内存里的键值数据库键是字符串值可以是你熟悉的字符串、哈希、列表、集合、有序集合甚至位图、流这些花活。正因为这样你得先搞清楚每种数据结构适合干什么再动手写代码不然很容易把Redis当普通map用平白浪费了它的能力。在正式开始之前先把最常用的连接方式写出来后面所有的增删改查示例都基于下面这套配置。我用的是redis-py这个库它是Python世界里最主流、维护也最积极的Redis客户端没有之一。安装很简单pip install redis就行。连接的时候我强烈建议加上decode_responsesTrue这个参数否则你读出来的数据全是bytes类型打印出来带个b前缀做字符串拼接和比较的时候会平白多出很多事。这个小细节几乎是所有新手都会遇到的第一个坑。1. 准备工作搞懂环境与连接里的几个关键点1.1 Redis服务怎么跑起来连接Redis之前你总得有个Redis服务端。本地开发的话最直接的方式是下载Redis源码编译或者用系统包管理器装一个。还有云厂商提供的托管实例界面点点就能创建省去维护的麻烦生产环境用托管实例也比较省心。我平时本地开发用的是docker跑一个实例docker run -d --name my-redis -p 6379:6379 redis:7-alpine这里把容器的6379端口映射到宿主机默认不带密码适合本机测试。加个--name方便管理以后开箱即用。要注意的是Windows下原生Redis一直没有官方支持但用Docker就没这个问题了这也是为什么我推荐开发环境用容器的原因。1.2 最基础的连接参数redis-py连接服务的入口是redis.Redis基本写法import redis r redis.Redis( host127.0.0.1, port6379, passwordNone, db0, decode_responsesTrue )host和port不用多说password如果服务端设置了访问密码就得填。重点说db和decode_responses。Redis默认有16个逻辑数据库编号从0到15通过flushdb清当前库、flushall清全部库。db这个参数就是选哪个库默认0就够了真要分业务请用不同key的前缀来区分别几个业务共用不同db到后期排查起来谁看谁迷糊。decode_responses我前面已经说了一定要设成True。它做的事情是让客户端在读取的时候自动把字节流解码成字符串。不设的话你写入的是hello读出来是bhello然后你拿去跟hello比较永远返回False写日志的时候还会莫名其妙多出个b排查半天才发现是这个原因。1.3 连接池的必要性很多教程只告诉你这么连但我实际项目里基本不会裸用redis.Redis而是先建一个连接池。为什么因为每次redis.Redis()都是新建连接高频请求下频繁创建和销毁TCP连接既慢又浪费资源。连接池相当于维护一批常驻连接用的时候借出用完归还省掉了反复握手的过程。pool redis.ConnectionPool( host127.0.0.1, port6379, db0, decode_responsesTrue, max_connections20 ) r redis.Redis(connection_poolpool)max_connections控制连接池里最多有多少个连接。这个数字不是越大越好量太大服务端文件描述符吃紧太小并发稍微一高就开始排队。一般单机应用给10到50就够用了具体要看你的QPS。如果再装个redis-py的哨兵模式或者集群模式连接池的思路也是一样的。1.4 连通性检查连接写好后用r.ping()验证一下能返回True说明网络、服务、认证全部正常。这一步能帮你快速定位是代码问题还是环境问题print(r.ping()) # True有个小技巧如果Redis实例做了游标分片或者通过代理访问ping也不一定能完全代表业务可用更稳妥的方式是set一个临时key再马上读出来把整条链路测一遍。2. 增删改查实操逐个击破五种核心数据类型2.1 字符串String最基础也最常用的类型String是Redis里最简单也是用得最多的类型适合存普通值、计数器、验证码这种场景。常用方法就这么几个# 增set如果key不存在就创建存在就覆盖 r.set(login:count, 1) # 查get返回字符串不存在返回None print(r.get(login:count)) # 1 # 改直接再set就是改append在末尾追加 r.append(login:count, 0) # 值变10 # 删delete可以删一个或多个key r.delete(login:count) # 设置过期时间单位秒 r.setex(sms:code:13800000000, 300, 123456)setex这个方法是“设置值同时指定过期时间”非常适合存短信验证码、临时token这种要自动失效的数据。它比先set再expire更可靠因为两步操作之间如果程序崩了key就可能永远不带过期时间。计数器也是String的招牌用法用incr和decrr.incr(page:view:home) r.incrby(page:view:home, 10) r.decr(stock:1001)incr是原子自增多进程并发下也不会重复计数这是Redis单线程模型带来的好处。这里强调一下在Python里你如果用get然后int再set两个进程同时操作就可能丢失一次更新而incr不会有这个问题。凡是需要计数的场景直接走incr就对了。2.2 Hash哈希描述一个对象的最优解Hash给我们的感觉就像Python里的字典一个key下面挂着一堆字段。存用户信息、商品信息、配置项这种结构化数据比用JSON字符串方便多了只取某个字段的时候不用整个解析出来。# 增hset如果字段不存在则新增存在则覆盖值 r.hset(user:1001, mapping{ name: 张三, age: 18, city: 上海 }) # 查hget取单个字段hgetall取全部 print(r.hget(user:1001, name)) # 张三 print(r.hgetall(user:1001)) # {name: 张三, age: 18, city: 上海} # 改hset再赋值就行hincrby可以做字段级别自增 r.hset(user:1001, age, 19) r.hincrby(user:1001, age, 1) # 删hdel删除一个或多个字段delete删除整个key r.hdel(user:1001, age) r.delete(user:1001) # 其他常用判断 print(r.hexists(user:1001, name)) # 字段是否存在 print(r.hlen(user:1001)) # 字段数量要注意的是hgetall会把所有字段一次性拉出来如果这个hash特别大对内存和网络都不友好。生产环境我更建议先用hkeys去看看有哪些字段或者只取真正需要的那几个字段。还有一个细节hset的旧版写法是一个一个字段传参新版redis-py推荐用mapping参数传字典干净利落。2.3 List列表队列和栈都是它Redis的List是一个双向链表支持从头部或者尾部插入和弹出。这个消息队列也好最新文章列表也好其实都是拿List实现的。# 增rpush从右边推入lpush从左边推入 r.rpush(queue:task, task-1) r.rpush(queue:task, task-2, task-3) r.lpush(queue:task, task-0) # 查lrange按索引范围取0和-1表示全部 print(r.lrange(queue:task, 0, -1)) # [task-0, task-1, task-2, task-3] # 改set按索引覆盖指定位置ltrim裁剪列表 r.lset(queue:task, 0, task-init) r.ltrim(queue:task, 0, 2) # 删lpop左边弹出rpop右边弹出lrem删除指定值 left r.lpop(queue:task) right r.rpop(queue:task) r.lrem(queue:task, 1, task-2) # count1表示最多删一个 # 长度 print(r.llen(queue:task))自己做队列的时候有一个比较著名的坑用lpop取任务业务处理一半程序崩了任务就丢了。可靠一点的做法是先用lrange看一眼处理完再用lrem删掉或者用Redis 5.0引入的Stream机制它带消费确认比List靠谱得多。轻量级内部任务用List没问题但核心业务队列我建议直接上Stream或者专业的消息中间件。2.4 Set集合去重和关系的利器Set是元素不重复的集合适合做标签、去重、共同好友这类场景。它的核心卖点就是集合运算比如交集、并集、差集一个命令就出来了在Python里 требуется好几行循环才能搞定。# 增sadd一次可以加多个 r.sadd(user:1001:fans, u1, u2, u3) r.sadd(user:1002:fans, u2, u3, u4) # 查smembers取全部sismember判断在不在 print(r.smembers(user:1001:fans)) # {u2, u3, u1} print(r.sismember(user:1001:fans, u1)) # True # 删srem移除指定值spop随机弹出一个 r.srem(user:1001:fans, u1) popped r.spop(user:1001:fans) # 集合运算 inter r.sinter(user:1001:fans, user:1002:fans) union r.sunion(user:1001:fans, user:1002:fans) diff r.sdiff(user:1001:fans, user:1002:fans) print(inter) # {u2, u3} # 数量 print(r.scard(user:1001:fans))这里是典型场景用户A和用户B的共同粉丝数据库表join可能要写半天Redis一行命令搞定。另外每天签到去重可以用sadd然后scard一数就是真实人数天然防重复。要注意smembers会把整个集合拉回来超大集合下慎用可以用sscan分批遍历。2.5 有序集合ZSet排行榜的标配ZSet每个元素都带一个分数scoreRedis会按分数排序。排行榜、延时队列、热门文章排序都是ZSet的主场。# 增zadd带分数可以一次加多个 r.zadd(rank:game, {playerA: 1000, playerB: 2000, playerC: 800}) # 查按排名取zrange从小到大zrevrange从大到小 print(r.zrevrange(rank:game, 0, 2, withscoresTrue)) # 分数最高的是playerB # 查单个分数和排名 print(r.zscore(rank:game, playerA)) # 1000.0 print(r.zrevrank(rank:game, playerA)) # 排名大的在前 # 改zincrby增加分数 r.zincrby(rank:game, 500, playerA) # 删zrem移除指定元素 r.zrem(rank:game, playerC) # 查数量与区间 print(r.zcard(rank:game)) print(r.zrangebyscore(rank:game, 1000, 2000))zincrby非常适合给排行榜动态加分并发下也不会错乱。延时队列的实现思路是把任务执行时间戳作为score然后循环用zrangebyscore取出到期的任务执行再zrem移除算是比较经典的轻量方案了。ZSet注意一点score是浮点类型如果要精确到分就存分为单位的整数避免浮点误差。3. 通用方法过期时间、序列化、管道与锁3.1 把key的有效期管理好Redis里最常用的通用方法就是expire和ttl。你设置缓存的时候通常都会给一个有效期不然key只增不减内存迟早爆掉。r.set(cache:hot, value) r.expire(cache:hot, 60) # 60秒后自动删除 print(r.ttl(cache:hot)) # 剩余秒数-1表示没有过期时间-2表示key不存在 r.persist(cache:hot) # 移除过期时间变为永久这里有一个常见的坑set一个已经存在的key并不会重置它的过期时间。我先set 60秒过期中间又set了一次同一个key结果还是原来的时间线不会重新计60秒。所以想“刷新缓存有效期”得再调一次expire。另外批量设置缓存的时候用setex会更顺手。缓存穿透和雪崩虽然听起来高级本质上就是对过期时间的两种误用。缓存穿透是指查一个不存在的key每次请求都打到数据库解决方法是哪怕空结果也缓存一段极短的时间或者用布隆过滤器挡一层。缓存雪崩是指大量key同一时间过期数据库瞬间压力大增解决思路是给过期时间加一个随机偏移比如expire(60 random.randint(0, 30))打散过期时间点。3.2 对象序列化怎么存刚才用Hash存对象的时候每个字段对应一个属性这当然是最优雅的方式。但有些时候你拿到的就是一段JSON字符串或者整体数据不需要按字段访问那直接存序列化后的字符串也完全可以。import json user {name: 张三, age: 18, city: 上海} r.set(user:2001, json.dumps(user, ensure_asciiFalse)) data json.loads(r.get(user:2001)) print(data[name])强调一下json.dumps里ensure_asciiFalse一定要带上否则中文会变成一行\uXXXX的转义序列可读性差不说还会多占空间。如果数据是Python自带的复杂结构有人喜欢用pickle但我个人不推荐跨服务使用pickle因为不同解释器版本可能不兼容而且pickle有安全风险千万别反序列化不可信来源的数据。3.3 管道Pipeline批量操作的性能武器如果业务里需要连续执行很多个Redis命令最直观的写法是循环调比如for i in range(100): r.set(fkey:{i}, i)这样每一轮都是一次网络往返100个key就是100次RTT。虽然Redis本身处理很快但网络IO时间反而成了瓶颈。用pipeline可以把命令先积压在客户端然后一次性发送给服务端执行再一次性收回结果性能提升非常明显。pipe r.pipeline() for i in range(100): pipe.set(fkey:{i}, i) pipe.execute()pipeline默认是非事务的只是批量发送想要事务可以用pipe r.pipeline(transactionTrue)这样所有命令会打包成一个事务执行中间不会被其他客户端的命令插队。需要注意pipeline虽然快也不是无脑用中间如果某个命令语法错误整体执行会受影响所以重要的写入操作建议还是一条条来或者做好错误处理。3.4 用Redis实现一个简单的分布式锁多进程抢资源的时候比如秒杀、定时任务唯一执行分布式锁是标配。Redis实现锁的核心是一条原子命令SET key value NX EX seconds。NX的意思只有key不存在时才设置成功EX表示同时设置过期时间这样可以避免两个动作分开执行导致锁失效的问题。redis-py里set方法支持这些参数# 尝试获取锁谁拿到谁就执行 ok r.set(lock:order:1001, owner, nxTrue, ex30) if ok: try: # 业务逻辑 pass finally: # 释放锁时最好先判断是自己持有的再删 lock_value r.get(lock:order:1001) if lock_value owner: r.delete(lock:order:1001)这里有个细节很多人会忽略释放锁的时候不建议无脑delete因为如果你的锁过期了别的进程已经拿到了新锁你去delete会把别人的锁删掉造成锁失效。所以释放前先get验证一下value是不是自己写进去的再删除。严谨一点还会配合Lua脚本保证“判断并删除”也是原子操作不过上面的写法在绝大多数业务场景下已经够用了。3.5 大key扫描用Scan别用Keys生产环境最危险的操作就是keys *它会阻塞Redis单线程导致所有请求排队严重的会让线上服务像卡死一样。要遍历key正确的是用scancursor 0 while True: cursor, keys r.scan(cursorcursor, count100) for key in keys: print(key) if cursor 0: breakscan是游标式遍历每次只取一部分key不会阻塞服务。它允许在遍历过程中key发生变化所以不保证返回的key一定是“当时完整快照”但对绝大多数运维场景来说完全够用。count控制的是单次扫描的大致数量。我自己在清理无用key、寻找大key的时候都是用scan配合type和debug object来做的从来没在生产环境敲过keys。4. 常见问题与排查技巧实录4.1 一图看懂高频报错与解决方案我把这些年自己还有团队里同事碰到的问题挑典型的整理成一张表遇到类似情况直接对号入座症状大概率原因解决办法ConnectionError: 127.0.0.1:6379Redis服务没启动或端口不对检查进程、docker容器状态netstat确认端口监听Authentication required服务端有密码但连接时没传password在Redis()里加password参数读出来的值是bxxx忘了写decode_responsesTrue连接池和Redis对象都加上get返回None但key明明存在操作的db库不对或者连了不同的实例确认db编号用select切换库复查程序偶尔报连接池超时连接数不够连接泄漏调大max_connections检查代码是否有close写入中文后读出来乱码序列化时没指定编码或JSON转义ensure_asciiFalse统一utf-8线上突然全部请求变慢有人在跑keys命令或大key操作立刻找到执行方换成scan定位大key拆分4.2 连接池耗尽这个坑我第一次在生产环境遇到ConnectionPool报错的时候还以为是Redis挂了结果服务端日志一切正常后来才发现是连接池的连接数被占满了。最典型的场景是你在循环里或者每次请求里创建新的redis.Redis对象而且是带着新连接池创建的用完没有释放连接越积越多最后池子耗尽。正确的做法是全局复用同一个连接池对象redis.Redis的底层连接自动归还并不需要你显式close。只有用底层Connection的时候才需要手动管理释放。排查命令很简单redis-cli info clients看connected_clients是不是堆得很高。如果应用侧没法快速定位还可以临时调大max_connections先顶上但根本解决还是找代码里创建连接的地方改成单例模式复用。4.3 内存爆掉的前兆与自救Redis是内存数据库内存就是它的容量边界。最常遇到的问题是缓存key一直不设过期时间或者过期时间设得太长数据只增不减。我的自查习惯是定期执行info memory看used_memory和maxmemory的比值一般超过70%就要开始警觉了。清理手段有几种给不重要的key分批设置较短过期时间用unlink异步删除大key比delete阻塞时间少得多配置maxmemory-policy allkeys-lru让Redis按LRU策略自动淘汰。这些都做过之后还是紧张就该考虑加内存或者换集群了。4.4 关于中文和二进制数据Redis的字符串存储本质是字节序列所以中文不存在“存不下”的问题问题往往出在序列化和反序列化环节。最常见的是json.dumps没设ensure_asciiFalse还有一种情况是数据库连接字符集和Redis存储时不一致把乱码带进来了。我的建议是所有写入Redis的文本统一用JSON格式、UTF-8编码读取时也统一按UTF-8处理别一个项目里既用GBK的日志来源又用UTF-8的接口入参早晚出问题。二进制文件也可以往Redis放但它不是文件系统放几十KB的小对象没问题大文件还是老老实实对象存储吧。4.5 数据一致性怎么保证更新数据库后更新Redis这是最常见的缓存策略但难点在于先更新谁、失败怎么办。我的习惯是能接受短暂不一致的业务优先用“先更新数据库再删除缓存”这种被动缓存因为删除操作幂等且失败的影响比写入错误数据小如果必须主动更新缓存就用消息队列异步同步两边的数据保证最终一致。还有一种做法是给缓存加版本号读取时发现版本过期就重新加载。没有银弹关键还是想清楚业务允许几秒钟的不一致。在我自己实操下来的体会是Python操作Redis真正的难点从来不在那几个API怎么调用而在于你怎么设计key、怎么设置过期时间、怎么避免阻塞、怎么保证高并发下不出错。上面这些代码和排查经验只要照着用基础场景基本就覆盖到了。最后再分享一个小习惯给所有key设计统一的前缀比如业务名:模块名:对象ID线上排查和定位问题能省一半时间。搞一个随手的临时key叫什么xxx回头自己都想不起来这种感觉你多经历几次自然就懂了。