functools.lru_cache不支持TTL机制,因其设计为纯LRU淘汰且无过期时间参数;需手写TTLCache类,用OrderedDict存(value, expire_time)并检查时间戳,注意LRU顺序更新、线程安全及精度权衡。
functools.lru_cache 不能直接加过期时间functools.lru_cache 是 Python 内置的 LRU 缓存,但它不支持 TTL(Time-To-Live)机制。缓存项一旦写入,就永远有效,直到被 LRU 淘汰或手动清除。你无法通过参数设置“5 秒后自动失效”。强行在函数内检查时间戳,会破坏装饰器的纯封装性,也容易漏掉并发访问下的竞态问题。
OrderedDict + 时间戳核心思路是:用 collections.OrderedDict 维护访问顺序,每个缓存值存储为 (value, expire_time) 元组;每次 get 前检查 expire_time 是否已过期,过期则删除并返回未命中。
关键实操点:
time.time()(非 time.monotonic())便于调试,但注意系统时间回拨会导致误删;生产环境可换用 time.monotonic() + 初始偏移OrderedDict.move_to_end(key) 必须在每次 get 成功后调用,否则 LRU 顺序错乱pop 再重插,避免残留旧过期时间maxsize)和 TTL 要正交处理:淘汰只看数量,过期只看时间示例片段(简化版):
from collections import OrderedDict import timeclass TTLCache: def init(self, maxsize=128, ttl=60): self.cache = OrderedDict() self.maxsize = maxsize self.ttl = ttl
def get(self, key): if key not in self.cache: return None value, expire_at = self.cache[key] if time.time() > expire_at: self.cache.pop(key) return None self.cache.move_to_end(key) # 更新 LRU 顺序 return value def put(self, key, value): if self.maxsize == 0: return if key in self.cache: self.cache.pop(key) elif len(self.cache) >= self.maxsize > 0: self.cache.popitem(last=False) # 弹出最久未用 self.cache[key] = (value, time.time() + self.ttl)用装饰器包装成类似
lru_cache的用法要复刻
@lru_cache(ttl=30)的体验,需支持带参装饰器、绑定到函数对象的独立缓存实例,并兼容cache_clear()等方法。注意点:
TTLCache 实例,避免跨函数污染cache_clear、cache_info 等方法挂到 wrapper 上,否则用户调用 func.cache_clear() 会报错typed=True(即不同类型的相同值视为不同 key),除非手动序列化类型信息
上面的 TTLCache 类默认不是线程安全的:get 和 put 中的多步操作(查、删、改、move)可能被并发打断。简单加 threading.Lock 会严重拖慢性能,尤其读多写少场景。
更实用的做法:
get)不加锁,允许短暂返回过期值(业务能容忍几毫秒偏差)put)加锁,保证 pop 和 set 原子性threading.RLock 并把整个 get 流程锁住——但请确认你的 QPS 是否真的需要真正难的不是实现,而是判断“过期”是否必须精确到毫秒级,以及能否接受缓存雪崩时的瞬时穿透。这些权衡点,比代码本身更影响最终效果。