Python与Redis操作文章目录Python与Redis操作前言一、环境准备与安装二、连接 Redis2.1 基本连接2.2 连接池三、常用数据类型操作3.1 字符串String3.2 哈希Hash3.3 列表List3.4 集合Set3.5 有序集合Sorted Set四、Pipeline 批量操作五、缓存实战六、分布式锁实现总结亮点总结适用场景扩展方向前言RedisRemote Dictionary Server是一款基于内存的高性能键值存储系统。凭借其丰富的数据结构字符串、哈希、列表、集合、有序集合等和亚毫秒级的响应时间Redis 被广泛应用于缓存、消息队列、排行榜、会话存储等场景。在面试中谈缓存时Redis 几乎是必问的话题——面试官不仅想知道你会不会SET和GET更想知道你是否理解缓存穿透、缓存雪崩等生产环境中的实际问题及解决方案。面试常见考点Redis的五种核心数据类型及其应用场景、连接池的作用和原理、Pipeline加速的原因减少网络RTT、缓存穿透/缓存雪崩的成因和防护、分布式锁的实现基于SET NX Lua脚本。在 Python 生态中redis-py是最流行的 Redis 客户端库提供了完整的命令映射和连接池管理。本文将带你从安装配置到高级应用全面掌握 Python 操作 Redis 的实战技巧。一、环境准备与安装pipinstallredis如果需要在异步环境中使用asyncio安装可选的异步支持pipinstallredis[hiredis]hiredis是一个用 C 编写的 Redis 协议解析器能显著提升解析速度。二、连接 Redis2.1 基本连接importredis# 方式一URL 连接rredis.from_url(redis://:passwordlocalhost:6379/0)# 方式二参数连接rredis.Redis(hostlocalhost,port6379,db0,passwordNone,decode_responsesTrue,# 自动解码为字符串而非 bytessocket_timeout5,socket_connect_timeout5,)# 测试连接try:r.ping()print(连接成功)exceptredis.ConnectionErrorase:print(f连接失败{e})2.2 连接池在高并发场景下每次请求都创建新连接效率很低。TCP三次握手和每次连接的网络开销是无法忽略的——在每秒千次请求的场景下连接建立和销毁的成本可能超过实际数据传输。ConnectionPool维护一组预先创建的长连接通过复用避免重复握手。连接池是典型的空间换时间策略用少量内存占用换取数十倍的延迟降低。importredis# 创建连接池poolredis.ConnectionPool(hostlocalhost,port6379,db0,max_connections50,decode_responsesTrue,)rredis.Redis(connection_poolpool)# 多线程环境下安全使用连接池fromconcurrent.futuresimportThreadPoolExecutordefworker(task_id):clientredis.Redis(connection_poolpool)client.set(ftask:{task_id},fprocessed-{task_id})valueclient.get(ftask:{task_id})returnvaluewithThreadPoolExecutor(max_workers10)asexecutor:resultslist(executor.map(worker,range(100)))print(f完成了{len(results)}个任务)注意连接池是线程安全的但多个线程之间不要共享同一个 Redis 客户端实例。推荐模式是每个线程从连接池创建独立的客户端。三、常用数据类型操作3.1 字符串String字符串是 Redis 最基本的数据类型一个 key 对应一个 value最大可存储 512MB# 基本读写r.set(username,张三)print(r.get(username))# 张三# 带过期时间秒r.setex(session:token,3600,abc123def)# 1小时后过期r.set(temp,data,ex300)# 5分钟后过期# 数值操作计数r.set(page:views,0)r.incr(page:views)# 1r.incrby(page:views,10)# 11r.decr(page:views)# 10# 仅当 key 不存在时设置分布式锁基础r.setnx(lock:order:10086,locked)3.2 哈希Hash哈希适合存储对象属性类似于 Python 字典# 存储用户信息r.hset(user:1001,mapping{name:李四,email:lisiexample.com,age:28,})# 获取所有字段userr.hgetall(user:1001)print(user)# {name: 李四, email: lisiexample.com, age: 28}# 单个字段读写r.hset(user:1001,score,95)print(r.hget(user:1001,score))# 95# 递增数值字段r.hincrby(user:1001,login_count,1)# 获取所有字段名或值print(r.hkeys(user:1001))print(r.hvals(user:1001))3.3 列表List列表是按插入顺序排序的字符串集合支持双端操作# 右侧入队队尾左侧出队队首 消息队列r.lpush(queue:tasks,task1,task2,task3)# 左侧添加taskr.rpop(queue:tasks)# 右侧弹出print(task)# task1# 范围获取r.lpush(logs,*[flog-{i}foriinrange(10)])recentr.lrange(logs,0,4)# 前5条print(recent)# 阻塞弹出消费者模式resultr.blpop(queue:tasks,timeout5)# 等待5秒3.4 集合Set集合是无序的、不重复的字符串集合# 标签系统r.sadd(tags:article:1,python,redis,caching)r.sadd(tags:article:2,python,mongodb)# 交集两篇文章的共同标签commonr.sinter(tags:article:1,tags:article:2)print(common)# {python}# 差集仅第一篇文章有的标签uniquer.sdiff(tags:article:1,tags:article:2)print(unique)# {redis, caching}# 判断成员print(r.sismember(tags:article:1,python))# True3.5 有序集合Sorted Set有序集合的每个成员对应一个分数score按分数排序# 排行榜r.zadd(leaderboard,{Alice:9500,Bob:8700,Charlie:9200,David:7600})# 按分数降序从高到低top3r.zrevrange(leaderboard,0,2,withscoresTrue)forname,scoreintop3:print(f{name}:{score})# 增加分数r.zincrby(leaderboard,300,Alice)# 获取排名从0开始print(r.zrevrank(leaderboard,Bob))# Bob 的排名四、Pipeline 批量操作当需要一次性执行多个命令时Pipeline可以将多个命令打包发送减少网络往返次数RTT。Pipeline的工作原理很简单客户端先将多个命令缓冲起来然后一次性发送给Redis服务器服务器依次执行后一起返回结果。这样100个命令只需要1次网络往返而不使用Pipeline则需要100次网络往返——在高延迟网络中这个差距可能达到数千倍。注意Pipeline不是事务——中间某个命令失败不会回滚之前的命令这和Redis的MULTI/EXEC事务不同。# 无 Pipeline每次命令一次网络往返importtime starttime.time()foriinrange(100):r.set(fpipe:key{i},fvalue{i})r.get(fpipe:key{i})print(f无 Pipeline{time.time()-start:.2f}s)# 使用 Pipelinestarttime.time()piper.pipeline()foriinrange(100):pipe.set(fpipe:key{i},fvalue{i})pipe.get(fpipe:key{i})resultspipe.execute()print(f有 Pipeline{time.time()-start:.2f}s)# 数十倍加速五、缓存实战以文章阅读量缓存为例子实现缓存穿透保护。缓存穿透是指查询一个不存在的数据如恶意请求不存在的文章ID由于缓存中也没有请求直接穿透到数据库。如果恶意请求大量不同ID就会对数据库造成巨大压力。解决方案就是缓存空值并设置短过期时间让数据库免受无效请求的反复冲击。缓存雪崩是指大量缓存同时过期所有请求瞬间打到数据库——这就像雪崩一样突然。随机化过期时间在基础TTL上叠加一个随机值能让缓存过期时间错峰避免了同时失效的问题。importjsonimporttimeclassArticleCache:def__init__(self,redis_client,db_client):self.rredis_client self.dbdb_clientdefget_article(self,article_id):获取文章优先从缓存获取cache_keyfarticle:{article_id}# 1. 尝试从缓存获取cachedself.r.get(cache_key)ifcached:returnjson.loads(cached)# 2. 缓存未命中从数据库查询articleself.db.query_article(article_id)ifarticleisNone:# 防止缓存穿透缓存空值设置短过期时间self.r.setex(cache_key,60,json.dumps(None))returnNone# 3. 写入缓存随机过期时间防止缓存雪崩ttl3600(article_id%600)# 1小时 0~10分钟随机self.r.setex(cache_key,ttl,json.dumps(article))returnarticledefincr_views(self,article_id):增加阅读量使用哈希todaytime.strftime(%Y%m%d)self.r.hincrby(farticle:{article_id}:stats,fviews:{today},1)self.r.expire(farticle:{article_id}:stats,86400*30)# 保留30天六、分布式锁实现Redis 的原子操作使其非常适合实现分布式锁。为什么需要分布式锁在单体应用中Python的threading.Lock可以保证线程安全但在多实例部署的微服务中threading.Lock只能锁住当前进程内的线程无法跨服务器协调。这时就需要一个中心化的锁机制——Redis凭借其单线程模型和原子操作成为了理想选择。实现分布式锁的标准公式SET lock_key uuid NX EX ttl。NXNot eXists确保只有一个客户端能成功获取锁EX设置过期时间防止死锁某个客户端获取锁后崩溃未释放。释放锁时使用Lua脚本确保检查身份和删除是原子的——如果不用Lua你检查通过后、删除之前的时间窗口内锁可能被另一个客户端获取导致你误删了别人的锁。importuuidclassRedisLock:def__init__(self,redis_client,lock_key,ttl10):self.rredis_client self.lock_keyflock:{lock_key}self.ttlttl self.lock_idstr(uuid.uuid4())defacquire(self,blockingTrue,timeoutNone):尝试获取锁returnself.r.set(self.lock_key,self.lock_id,nxTrue,exself.ttl)defrelease(self):释放锁使用 Lua 脚本保证原子性lua_script if redis.call(GET, KEYS[1]) ARGV[1] then return redis.call(DEL, KEYS[1]) else return 0 end returnself.r.eval(lua_script,1,self.lock_key,self.lock_id)# 使用示例lockRedisLock(r,order:10086,ttl30)iflock.acquire():try:# 执行临界区代码print(处理订单...)finally:lock.release()总结本文介绍了 Python 通过 redis-py 操作 Redis 的核心技术连接池提升高并发下的连接效率五种核心数据类型覆盖了缓存、队列、排行榜等典型场景Pipeline显著减少网络开销分布式锁则为分布式系统提供了协调能力。在实际项目中Redis 通常作为缓存加速层存在合理设计缓存策略如过期时间随机化、空值保护能有效防止缓存雪崩和穿透。掌握这些技巧你就能将 Redis 的能力充分发挥到 Python 应用中。亮点总结五种核心数据类型全覆盖String计数器/分布式锁基础、Hash对象存储、List消息队列/栈、Set标签系统/交集运算、Sorted Set排行榜每种类型都配有实战代码示例。连接池与高并发ConnectionPool 实现连接复用避免频繁握手开销强调线程安全的正确使用模式每线程独立客户端适合高并发场景。Pipeline 批量操作通过将多个命令打包为一次网络往返数十倍提升批量读写效率是 Redis 性能优化的核心技巧。缓存穿透与雪崩防护空值缓存 短过期时间防止穿透随机 TTL 防止雪崩体现了生产级缓存设计的工程思维。分布式锁实现基于 SET NX Lua 脚本原子释放的分布式锁方案完整展示了获取与释放的全流程。适用场景Web 应用缓存加速将热点数据用户会话、文章内容、配置信息缓存到 Redis大幅降低数据库查询压力提升接口响应速度。排行榜与计数器系统利用 Sorted Set 的按分数排序和 INCR 原子递增轻松实现实时排行榜、阅读量统计和限流计数。分布式锁与任务去重在多实例部署的服务中通过 Redis 锁协调对共享资源的互斥访问或防止重复处理同一消息。扩展方向Redis 集群与哨兵模式从单机 Redis 过渡到高可用架构学习 Sentinel哨兵实现故障转移以及 Redis Cluster 实现数据分片。Python 性能优化技巧本系列第 96 篇将 Redis 缓存与 Pipeline 批量操作结合进一步优化系统瓶颈点形成缓存 计算优化的双重加速方案。Celery 任务队列与 RedisRedis 作为 Celery 的 broker 和 result backend实现异步任务的分发与结果存储构建可靠的分布式任务处理系统。