因为@lru_cache不支持自动TTL过期,需手动清理易致脏数据;cachetools.TTLCache每次访问自动校验过期,支持线程安全、单调时钟和动态配置,更可靠。

为什么不用 @lru_cache 而选 cachetools.TTLCache
@lru_cache 默认不支持 TTL(Time-To-Live),过期时间得靠手动清空或重写逻辑,实际业务中容易漏掉清理、导致脏数据;cachetools.TTLCache 在每次 get 或 __contains__ 时自动检查过期,更可靠。它还支持线程安全(需传 maxsize 且非 None)、可定制 ttl 和 timer(比如用 time.monotonic 避免系统时间回拨问题)。
常见错误现象:用 @lru_cache(maxsize=128) + time.time() 手动判断时间戳,但没在每次访问时校验,缓存项永远不淘汰;或者忘了加锁,在多线程下调用 cache.clear() 误伤其他 key。
- 必须显式传
maxsize(哪怕设为128),否则TTLCache内部不会启用过期检查逻辑 - 默认
timer=time.time,若服务可能跨时区或系统时间被 NTP 调整,建议换timer=time.monotonic - 不支持异步函数装饰——如果被缓存的是
async def,得用cached(cache=...)+ 手动 await 包装,不能直接@cached
怎么初始化一个线程安全的 TTL 进程级缓存
进程级意味着所有线程共享同一份缓存实例,所以初始化一次、全局复用即可。关键点是:开 maxsize、关 typed(除非真需要区分 int(1) 和 float(1.0))、配好 timer。
import time from cachetools import TTLCache <h1>推荐初始化方式:线程安全 + 单调时钟 + 显式大小</h1><p>cache = TTLCache( maxsize=512, ttl=300, # 5 分钟 timer=time.monotonic )
-
maxsize=0等价于无限制,但此时 TTL 检查仍有效;不过不推荐,内存可能无限增长 - 不要在函数内重复创建
TTLCache()实例,那只是局部变量,起不到进程级共享作用 - 如果用在 FastAPI/Flask 的依赖注入里,确保它是模块级全局变量,或通过
lifespan注册单例
怎么安全地读写缓存(避免 KeyError 和竞态)
TTLCache 本身不是 dict 子类,但支持 __getitem__、__setitem__、get()。但直接用 cache[key] 在 key 不存在或已过期时会抛 KeyError;而 cache.get(key) 返回 None(即使值本身是 None),无法区分“未命中”和“存了 None”。更稳妥的做法是始终用 get + 显式判断,并配合 setdefault 避免重复计算。
快速生成专业的 Python 脚本和应用代码。一键创建完整项目结构,支持CLI、API、爬虫、Bot、Django等多种项目类型,包含完整的项目结构、配置文件、依赖管理、测试、README和文档。
立即学习“Python免费学习笔记(深入)”;
# 安全读取:先 get,再按需计算并 set
def get_user_profile(user_id: int):
key = f"user:{user_id}"
value = cache.get(key)
if value is not None:
return value
# 计算并写入(注意:这里没锁,高并发下可能重复计算一次,可接受)
value = _fetch_from_db(user_id)
cache[key] = value
return value
<h1>或用 setdefault(内部有锁,但仅限 cache 实例启用了 maxsize)</h1><p>value = cache.setdefault(key, _fetch_from_db(user_id))
-
setdefault是原子操作,但只在maxsize > 0时才真正加锁;maxsize=None下它退化为普通 get+set,无锁 - 不要用
cache[key] = ...后立刻cache[key]判断是否写入成功——过期间隔内可能已被淘汰 - 如果业务要求强一致性(比如金融类场景),缓存层应配合版本号或更新时间戳,TTLCache 本身不提供 CAS 能力
怎么验证缓存是否真的在淘汰、有没有内存泄漏
最直接的方式是观察 cache.currsize 和 len(cache)(二者等价),配合定时日志;也可以用 cache.popitem() 手动触发淘汰看行为。注意:currsize 只统计未过期项,过期项会在下次访问时惰性清理,所以刚过期时 currsize 不会立刻下降。
# 日志示例:每分钟打一次缓存状态
import logging
logging.info(f"cache size={cache.currsize}, hits={cache.hits}, misses={cache.misses}")
-
cache.hits/cache.misses需要初始化时加getsizeof=lambda x: 1才开启统计(默认关闭) - 用
objgraph查看TTLCache实例引用链,确认没有意外闭包持有了大对象(比如把整个 request 对象塞进缓存) - 如果发现
currsize持续上涨不降,大概率是ttl设得过大,或timer被篡改(比如混用了time.time和time.monotonic)
TTLCache 的淘汰不是后台线程驱动的,而是访问驱动的——没人查,过期项就一直占着内存。这点和 Redis 的被动+主动混合淘汰不同,生产环境要注意低频 key 的内存滞留风险。

















