redis连接池需在应用启动时初始化复用,避免路由中反复创建;@cache.cached()的timeout是懒刷新窗口而非ttl;缓存键须带环境与服务前缀;cache.get()返回none不等于未命中,需显式过滤空值。

redis-py连接池必须初始化一次,别在路由里反复new
直接在视图函数里写 redis.Redis(host="localhost"),跑几天就可能触发 ConnectionRefusedError 或大量 TimeoutError。Flask 多 worker 场景下,每个请求都新建连接,系统文件描述符很快耗尽。
正确做法是在应用启动时创建连接池,复用给所有请求:
- 用
redis.ConnectionPool初始化,参数建议:max_connections=20、socket_timeout=1、retry_on_timeout=True - 传给
redis.Redis(connection_pool=pool),而不是 host/port 等参数 - 挂到
app.config["REDIS_CLIENT"]或用 Flask 扩展方式管理,避免全局变量或g对象误用
@cache.cached() 的 timeout 不是 TTL,而是懒刷新窗口
@cache.cached(timeout=300) 不代表“5 分钟后自动失效”,而是“首次缓存后,接下来 5 分钟内所有请求都返回旧值;第 301 秒的请求会阻塞执行原函数,并把新结果覆盖缓存”。高并发下容易引发雪崩——多个请求同时穿透缓存查 DB。
应对策略:
Redis 缓存和数据结构管理技能。通过自然语言操作 Redis,支持 String、Hash、List、Set、ZSet、Stream 等数据结构操作。当用户提到 Redis、缓存、消息队列、会话存储时使用此技能。
- 对关键接口加
refresh=True(需 Flask-Caching ≥ 2.0.0),启用后台异步刷新 - 手动预热:用
cache.set(key, value, timeout)+ 定时任务,在低峰期主动写入 - timeout 值要贴合业务更新频率,比如用户列表每 3 分钟变一次,就别设 3600
缓存键必须带环境+服务前缀,别信 keys *
用 f"user:{user_id}" 这种裸键名上线,线上出问题根本分不清是哪个服务、哪个环境、哪个分支写的,还容易被其他模块覆盖。更糟的是,有人用 keys user:* 查键,这在生产 Redis 上是阻塞操作,会卡住整个实例。
键设计规范:
- 统一格式:
f"{APP_NAME}:{ENV}:user:{user_id}",例如"myflask:prod:user:123" - 禁止在键里拼 JSON 或复杂结构——Redis 键只支持字符串,序列化后不可读、难调试
- 查键用
scan:命令行执行redis-cli --scan --pattern "myflask:prod:user:*",代码里用client.scan(match="myflask:prod:user:*")
cache.get() 返回 None 不等于没命中,空值要显式过滤
很多人写 if not cache.get(key): do_db_query(),结果数据库被打爆——因为 redis-py 把 None、""、b"" 都当有效值存进去了,而 get() 只有查不到键时才真返回 None。你无法区分“缓存里存了个空值”和“压根没这个键”。
安全写法:
- 写缓存前过滤:
if result is not None and result != "": cache.set(key, result, ex=300) - 读缓存时辅助判断:
cache.exists(key)(注意它和get()非原子,仅用于排查) - 更稳妥:约定空值存为特殊标记,如
"__NULL__",读到就当未命中
incr/decr 原生命令,别绕路 Python 运算再 set;用户资料更新后,别只 delete,要 set 新值,或者加锁防止并发重复查库。Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!










