上个月帮一个创业团队排查线上事故他们的电商活动页在大促高峰期整整卡了十分钟。监控显示数据库连接数直接打满慢查询堆积了上千条。创始人盯着屏幕问我“不就是展示一下商品详情吗怎么就把数据库干崩了”我看了一眼代码每次请求都直连MySQL查商品信息热门商品被上千人同时刷数据库扛得住才怪。其实这个问题有个标准解法——缓存。Redis作为业界主流的内存数据库配合Python的redis-py库能在不改变业务代码结构的前提下把数据库的查询压力降低90%以上。今天我们就从零开始聊聊怎么用Python操作Redis搭一套真正能打的缓存系统。基础篇先让Redis跑起来安装redis-py只需要一行命令pip install redis连接Redis的代码也极其简单import redis # 连接本地Redis r redis.Redis(hostlocalhost, port6379, db0, decode_responsesTrue) # 测试连接 r.set(foo, bar) print(r.get(foo)) # 输出: bar这里有个小细节decode_responsesTrue会让返回的结果自动从字节串转成字符串省去手动decode的麻烦。如果是生产环境建议用连接池管理连接避免频繁创建销毁消耗资源pool redis.ConnectionPool(hostlocalhost, port6379, db0, max_connections10) r redis.Redis(connection_poolpool)实战篇缓存最简单的写法最常见的缓存场景是数据库查询。一个用户信息服务如果不加缓存代码长这样def get_user(user_id): # 直接查数据库 return db.query(fSELECT * FROM users WHERE id{user_id})加一层Redis缓存代码变成这样def get_user(user_id): # 先查缓存 cache_key fuser:{user_id} user r.get(cache_key) if user: return json.loads(user) # 缓存命中直接返回 # 缓存未命中查数据库 user db.query(fSELECT * FROM users WHERE id{user_id}) # 写入缓存设置过期时间 r.setex(cache_key, 3600, json.dumps(user)) return user这个模式叫Cache-Aside是业界最通用的缓存策略。流程很简单读的时候先读缓存没有就查数据库然后回写写的时候先更新数据库然后删除缓存或者更新缓存。这里有两个关键点。一是缓存要有过期时间。上面的setex设置了3600秒避免缓存项永远驻留导致数据不一致。二是key的命名规范。用user:1001这样的格式冒号分隔不同部分在Redis里会自动按层级展示调试时一目了然。进阶篇用装饰器把缓存写成一行上面的写法已经能解决问题但还是不够优雅。每次都要手写缓存key、手动序列化、手动处理异常重复代码太多。Python的装饰器可以把这些脏活累活封装起来。一个最简版缓存装饰器可以这么写from functools import wraps import json def redis_cache(ttl300): def decorator(func): wraps(func) def wrapper(*args, **kwargs): # 生成缓存key函数名 参数 key f{func.__name__}:{str(args)}:{str(kwargs)} # 尝试从缓存获取 cached r.get(key) if cached: return json.loads(cached) # 执行原函数 result func(*args, **kwargs) # 写入缓存 r.setex(key, ttl, json.dumps(result)) return result return wrapper return decorator # 使用 redis_cache(ttl600) def get_user(user_id): return db.query(fSELECT * FROM users WHERE id{user_id})这样一来业务代码完全不需要关心缓存逻辑一个装饰器搞定。实际项目中可以用更成熟的库比如redis_func_cache它支持LRU、LFU等多种淘汰策略还封装好了Lua脚本保证原子性。坑点篇缓存穿透、击穿、雪崩怎么破缓存用不好有时候比不用还糟糕。三个经典问题值得留意。缓存穿透指查询一个根本不存在的数据。每次请求都绕过缓存直击数据库如果被恶意利用数据库分分钟被打挂。解决方案是缓存空值def get_user(user_id): user r.get(fuser:{user_id}) if user is not None: # 注意None表示缓存未命中空字符串表示缓存了空值 return user if user ! NULL else None user db.query(...) # 无论查没查到都写缓存 r.setex(fuser:{user_id}, 600, user or NULL) return user缓存击穿指某个热点key过期瞬间大量并发请求同时穿透到数据库。用分布式锁可以解决def get_hot_data(key): data r.get(key) if data: return data # 加锁只允许一个线程去查数据库 with r.lock(flock:{key}, timeout10): # 双重检查拿到锁后可能已经被其他线程更新了 data r.get(key) if data: return data data expensive_query() r.setex(key, 3600, data) return data缓存雪崩指大量key同时过期导致数据库瞬时压力暴增。解决方案是给过期时间加随机偏移量import random # 基础过期时间3600秒加上0-300秒的随机偏移 expire 3600 random.randint(0, 300) r.setex(key, expire, value)高阶篇多级缓存让速度再翻倍单靠Redis做缓存每次请求还是有一次网络开销。如果能把最热的数据放在应用本地内存里速度能再快一个数量级。这就是多级缓存架构本地缓存毫秒级→ Redis集群亚毫秒级→ 数据库毫秒级。80%的请求被本地缓存拦截剩下的20%由Redis承载数据库几乎只处理写请求和缓存未命中的场景。redis-py自带了本地缓存模块_LocalCache可以搭配使用from redis._cache import _LocalCache, EvictionPolicy # 初始化本地缓存最多存10000条30秒过期LRU淘汰策略 local_cache _LocalCache(max_size10000, ttl30, eviction_policyEvictionPolicy.LRU) def get_user_with_multilevel_cache(user_id): # 构造命令元组作为缓存key command (GET, fuser:{user_id}) # 查本地缓存 cached local_cache.get(command) if cached: return cached # 查Redis user r.get(fuser:{user_id}) if user: # 写入本地缓存 local_cache.set(command, user, keys_in_command[fuser:{user_id}]) return user # 查数据库 user db.query(...) r.setex(fuser:{user_id}, 3600, user) return user这套架构在实践中有几个优化点热点数据可以提前预热比如活动开始前把商品信息加载到缓存监控指标要跟上重点关注本地缓存命中率目标90%以上和Redis查询延迟目标1ms以下数据更新时要同时淘汰两级缓存保证一致性。收尾回到开头那个创业团队的故事。后来帮他们把用户信息和商品详情都加了Redis缓存数据库连接数从打满降到个位数接口响应时间从秒级降到几十毫秒。技术负责人发了条朋友圈“原来我们之前一直在用石器时代的方式写代码。”Redis缓存的本质很简单——用内存换速度用空间换时间。但用好它需要理解背后的数据一致性、过期策略、并发控制这些细节。希望这篇文章能帮你把这些细节串起来写出真正高效的缓存代码。
Python操作Redis：高效缓存设计与实战
上个月帮一个创业团队排查线上事故他们的电商活动页在大促高峰期整整卡了十分钟。监控显示数据库连接数直接打满慢查询堆积了上千条。创始人盯着屏幕问我“不就是展示一下商品详情吗怎么就把数据库干崩了”我看了一眼代码每次请求都直连MySQL查商品信息热门商品被上千人同时刷数据库扛得住才怪。其实这个问题有个标准解法——缓存。Redis作为业界主流的内存数据库配合Python的redis-py库能在不改变业务代码结构的前提下把数据库的查询压力降低90%以上。今天我们就从零开始聊聊怎么用Python操作Redis搭一套真正能打的缓存系统。基础篇先让Redis跑起来安装redis-py只需要一行命令pip install redis连接Redis的代码也极其简单import redis # 连接本地Redis r redis.Redis(hostlocalhost, port6379, db0, decode_responsesTrue) # 测试连接 r.set(foo, bar) print(r.get(foo)) # 输出: bar这里有个小细节decode_responsesTrue会让返回的结果自动从字节串转成字符串省去手动decode的麻烦。如果是生产环境建议用连接池管理连接避免频繁创建销毁消耗资源pool redis.ConnectionPool(hostlocalhost, port6379, db0, max_connections10) r redis.Redis(connection_poolpool)实战篇缓存最简单的写法最常见的缓存场景是数据库查询。一个用户信息服务如果不加缓存代码长这样def get_user(user_id): # 直接查数据库 return db.query(fSELECT * FROM users WHERE id{user_id})加一层Redis缓存代码变成这样def get_user(user_id): # 先查缓存 cache_key fuser:{user_id} user r.get(cache_key) if user: return json.loads(user) # 缓存命中直接返回 # 缓存未命中查数据库 user db.query(fSELECT * FROM users WHERE id{user_id}) # 写入缓存设置过期时间 r.setex(cache_key, 3600, json.dumps(user)) return user这个模式叫Cache-Aside是业界最通用的缓存策略。流程很简单读的时候先读缓存没有就查数据库然后回写写的时候先更新数据库然后删除缓存或者更新缓存。这里有两个关键点。一是缓存要有过期时间。上面的setex设置了3600秒避免缓存项永远驻留导致数据不一致。二是key的命名规范。用user:1001这样的格式冒号分隔不同部分在Redis里会自动按层级展示调试时一目了然。进阶篇用装饰器把缓存写成一行上面的写法已经能解决问题但还是不够优雅。每次都要手写缓存key、手动序列化、手动处理异常重复代码太多。Python的装饰器可以把这些脏活累活封装起来。一个最简版缓存装饰器可以这么写from functools import wraps import json def redis_cache(ttl300): def decorator(func): wraps(func) def wrapper(*args, **kwargs): # 生成缓存key函数名 参数 key f{func.__name__}:{str(args)}:{str(kwargs)} # 尝试从缓存获取 cached r.get(key) if cached: return json.loads(cached) # 执行原函数 result func(*args, **kwargs) # 写入缓存 r.setex(key, ttl, json.dumps(result)) return result return wrapper return decorator # 使用 redis_cache(ttl600) def get_user(user_id): return db.query(fSELECT * FROM users WHERE id{user_id})这样一来业务代码完全不需要关心缓存逻辑一个装饰器搞定。实际项目中可以用更成熟的库比如redis_func_cache它支持LRU、LFU等多种淘汰策略还封装好了Lua脚本保证原子性。坑点篇缓存穿透、击穿、雪崩怎么破缓存用不好有时候比不用还糟糕。三个经典问题值得留意。缓存穿透指查询一个根本不存在的数据。每次请求都绕过缓存直击数据库如果被恶意利用数据库分分钟被打挂。解决方案是缓存空值def get_user(user_id): user r.get(fuser:{user_id}) if user is not None: # 注意None表示缓存未命中空字符串表示缓存了空值 return user if user ! NULL else None user db.query(...) # 无论查没查到都写缓存 r.setex(fuser:{user_id}, 600, user or NULL) return user缓存击穿指某个热点key过期瞬间大量并发请求同时穿透到数据库。用分布式锁可以解决def get_hot_data(key): data r.get(key) if data: return data # 加锁只允许一个线程去查数据库 with r.lock(flock:{key}, timeout10): # 双重检查拿到锁后可能已经被其他线程更新了 data r.get(key) if data: return data data expensive_query() r.setex(key, 3600, data) return data缓存雪崩指大量key同时过期导致数据库瞬时压力暴增。解决方案是给过期时间加随机偏移量import random # 基础过期时间3600秒加上0-300秒的随机偏移 expire 3600 random.randint(0, 300) r.setex(key, expire, value)高阶篇多级缓存让速度再翻倍单靠Redis做缓存每次请求还是有一次网络开销。如果能把最热的数据放在应用本地内存里速度能再快一个数量级。这就是多级缓存架构本地缓存毫秒级→ Redis集群亚毫秒级→ 数据库毫秒级。80%的请求被本地缓存拦截剩下的20%由Redis承载数据库几乎只处理写请求和缓存未命中的场景。redis-py自带了本地缓存模块_LocalCache可以搭配使用from redis._cache import _LocalCache, EvictionPolicy # 初始化本地缓存最多存10000条30秒过期LRU淘汰策略 local_cache _LocalCache(max_size10000, ttl30, eviction_policyEvictionPolicy.LRU) def get_user_with_multilevel_cache(user_id): # 构造命令元组作为缓存key command (GET, fuser:{user_id}) # 查本地缓存 cached local_cache.get(command) if cached: return cached # 查Redis user r.get(fuser:{user_id}) if user: # 写入本地缓存 local_cache.set(command, user, keys_in_command[fuser:{user_id}]) return user # 查数据库 user db.query(...) r.setex(fuser:{user_id}, 3600, user) return user这套架构在实践中有几个优化点热点数据可以提前预热比如活动开始前把商品信息加载到缓存监控指标要跟上重点关注本地缓存命中率目标90%以上和Redis查询延迟目标1ms以下数据更新时要同时淘汰两级缓存保证一致性。收尾回到开头那个创业团队的故事。后来帮他们把用户信息和商品详情都加了Redis缓存数据库连接数从打满降到个位数接口响应时间从秒级降到几十毫秒。技术负责人发了条朋友圈“原来我们之前一直在用石器时代的方式写代码。”Redis缓存的本质很简单——用内存换速度用空间换时间。但用好它需要理解背后的数据一致性、过期策略、并发控制这些细节。希望这篇文章能帮你把这些细节串起来写出真正高效的缓存代码。
