1. 从“重复计算”说起为什么我决定手写一个带过期时间的记忆化装饰器做 Python 开发这些年我数不清遇到过多少次“同一个函数在短时间内被疯狂重复调用”的场景。最典型的就是数据库查询接口、外部 API 请求封装、配置信息读取这一类操作。前端点一次页面后端就要查好几次数据库查的还是同一张表、同一个 id数据在半秒钟内根本不会变但每次调用都老老实实重新查一遍。这种浪费在开发环境里感觉不明显一旦上了生产、并发一上来数据库连接数直接被打满接口响应时间从 10ms 飘到 500ms全靠数据库硬扛。Python 内置的functools.lru_cache确实能解决“重复计算”的问题它本质上就是一个记忆化装饰器把函数的输入参数映射到输出结果下次调用同一个参数时直接返回缓存值。但lru_cache有个很尴尬的缺陷——它没有“过期时间”的概念。缓存一旦写入要么一直存活到进程退出要么等缓存条目超过maxsize被 LRU 算法挤掉。在数据会变化的场景里这种“永远有效”的缓存就是定时炸弹。比如我缓存了一个用户的积分余额结果用户消费完了积分我这边还傻乎乎地把旧数据返回给客户端用户看到余额没变投诉就来了。我不否认lru_cache的价值它简单、轻量、零配置适合缓存那些“永久不变”的计算结果。但真实业务里绝大多数数据都有一个“有效期”配置项的生效周期可能是 5 分钟库存数量可能每 10 秒就要刷新一次Token 的过期时间可能只有 1 小时。这种场景下一个带 TTLTime To Live的缓存装饰器才是正解。市面上当然有现成的第三方库比如cachetools里面的TTLCache开箱即用。但用别人的库和“从零手写”完全是两个学习路径。手写的好处是你能真正理解缓存背后的原理缓存键是怎么生成的、过期时间是怎么判断的、并发环境下怎么保证不缓存错数据、占用的内存怎么控制。而且很多公司出于安全和依赖管理的考虑不允许随便引入第三方库自己写一个足够健壮的装饰器20 行代码就能覆盖 80% 的需求比引入一个库划算得多。这篇文章我就完整记录一下我的实现过程。从最基础的记忆化装饰器开始逐步加入过期时间、并发保护、内存上限、主动刷新这些功能最后做一轮对比测试把实际收益量化出来。整个代码我都会贴出来你可以直接复制到自己的项目里改一改就能用。2. 基础认知先行弄清楚缓存到底在解决什么问题2.1 缓存命中与未命中的执行路径在动手写代码之前我先把缓存的运行机制理清楚。一个带过期时间的记忆化装饰器本质上是在原本的函数调用前加了一层“检查站”。调用发生时装饰器先检查缓存里有没有对应参数的结果以及这个结果有没有过期。如果缓存命中且未过期直接把之前存好的结果返回函数体根本不执行如果缓存未命中或者已经过期才真正执行函数拿到结果后重新写入缓存再返回给调用方。这里有一个关键细节很多第一次接触缓存的人会误以为“过期时间”是从函数被调用的那一刻开始倒计时其实不是。更合理的做法是“从结果写入缓存的那一刻开始计时”。也就是说我 10:00:00 执行了一次函数把结果存进缓存过期时间设为 60 秒那么这个缓存应该在 10:01:00 失效而不是在我下一次调用比如 10:00:30之后再等 60 秒。后者的实现会显著延长缓存的实际存活时间数据新鲜度完全不可控。我写了一个简单的流程伪代码帮助理解def cached_call(args): 构造缓存键 key 尝试从缓存字典读取 (value, expire_at) 如果 key 存在 并且 expire_at 当前时间: 返回 value 否则: value 真正的函数执行(args) 将 (value, now ttl) 写入缓存 返回 value这个流程是后面所有代码的骨架不管装饰器加多少功能核心逻辑不会变。2.2 为什么functools.lru_cache不够用functools.lru_cache底层用的是哈希表加双向链表实现 LRU 淘汰性能毋庸置疑。但它缺的就是“时间维度”。我举个例子你就明白了from functools import lru_cache import time lru_cache(maxsize128) def get_user_level(user_id: int) - int: # 模拟耗时查询 time.sleep(0.5) return user_id % 5 print(get_user_level(1)) # 耗时 0.5 秒 print(get_user_level(1)) # 直接返回缓存耗时几乎为 0这段代码的问题在于第二次调用虽然很快但如果用户的等级在两次调用之间发生了变化比如后台管理员调整了用户等级第二次调用返回的就是过期的脏数据。lru_cache给了我们一个幻觉它让我们以为“所有重复计算都免费了”但现实是很多计算结果会随时间失效没有任何缓存策略能仅凭 LRU 淘汰就保证数据的时效性。还有一点容易被忽略lru_cache会强引用函数的参数和返回值。如果你缓存的对象是大型不可变数据比如几十 MB 的文本即使设置了maxsize1那一个缓存条目也会一直占着内存直到被其他 key 挤掉。对于需要精确控制缓存生命周期的场景这显然不够用。2.3 装饰器、闭包与缓存字典之间的关系写记忆化装饰器绕不开 Python 的闭包机制。装饰器实际上是一个接收函数作为参数、返回新函数的“工厂函数”。新函数通过闭包捕获了装饰器内部创建的缓存字典从而在每次调用时都能访问到同一份缓存数据。理解这一点的关键是每次用同一个装饰器去装饰不同的函数都会生成独立的闭包环境、独立的缓存字典。我用两种方式对比一下# 方式一同一个装饰器装饰两个函数 def ttl_cache(ttl: int): def decorator(func): cache {} def wrapper(*args, **kwargs): # 对 func 的缓存逻辑 ... return wrapper return decorator ttl_cache(ttl30) def get_order_count(user_id: int): ... ttl_cache(ttl30) def get_points(user_id: int): ...get_order_count和get_points各有各的cache字典互不干扰。这是装饰器的基本特性但在实际项目中我遇到过有人图省事把缓存字典定义在模块级别、所有函数共用一个 dict结果不同函数的缓存 key 互相冲突排查了半天才发现是缓存串了。这是一个很重要的设计红线缓存必须跟被装饰的函数绑定而不是全局共享。3. 从零搭建第一阶段最简单的记忆化装饰器3.1 缓存键的构造——细节决定了你会不会踩坑写记忆化装饰器第一个要解决的问题就是“怎么唯一标识一次函数调用”。我的方案是用函数名加上参数值构造一个 hash 值。这里要注意几个坑第一个坑是参数顺序。f(a1, b2)和f(b2, a1)在 Python 语义中是等价的但如果我把参数直接拼成字符串会得到两个不同的 key导致缓存失效、重复计算。所以必须把关键字参数按 key 排序之后再拼。第二个坑是参数类型。1和1.0在 Python 里如果用比较是相等的但如果我把参数直接转成字符串str(1)和str(1.0)又不一样。比较严谨的做法是用repr()加类型信息或者直接把参数打包进一个数据结构最后统一做 hash。第三个坑是参数不可哈希。如果传入的是 list、dict 这类可变对象直接做 hash 会抛TypeError。实际业务里这种场景不常见但一旦出现整个程序就崩了。稳妥的做法是捕获异常遇到不可哈希的参数直接不缓存。下面是我在实际项目中使用的缓存键构造方式import hashlib import inspect def make_cache_key(func, args, kwargs): # 将参数统一转换成可哈希的表示 key_parts [func.__module__, func.__qualname__] key_parts.extend(repr(arg) for arg in args) key_parts.extend(f{k}{repr(v)} for k, v in sorted(kwargs.items())) raw |.join(key_parts) return hashlib.sha256(raw.encode(utf-8)).hexdigest()用sha256而不是直接用hash()的原因是hash()的算法在 Python 进程重启后会随机化同一函数同一次调用在两次进程运行中会产生不同的 key。虽然大多数场景下缓存是进程内的、生命周期只有一次但如果哪天你想把缓存序列化到 Redis 或者文件里用hash()就会有坑。用sha256虽然慢一点但稳定、可迁移、冲突概率极低。3.2 基础版本代码与执行逻辑分析第一版我实现得尽量简洁只做记忆化不加过期时间import functools import hashlib import inspect import time def memorize(func): cache {} functools.wraps(func) def wrapper(*args, **kwargs): key make_cache_key(func, args, kwargs) if key in cache: return cache[key] result func(*args, **kwargs) cache[key] result return result return wrapper这个版本的执行流程很直白先查 key 在不在缓存字典里在就直接返回不在就执行函数体、存缓存、返回值。你可以把它挂到任何纯函数上试试效果性能提升非常明显尤其是那些内部有循环、递归或者数据库查询的函数。但到了这版我意识到一个严重的问题缓存永远不过期。如果被装饰的函数返回的是会变化的数据这个装饰器反而有害。所以我必须要加入时间维度。4. 核心升级让缓存学会“过期”4.1 TTL 参数设计——从装饰器参数的三种写法说起给装饰器加 TTL 参数有几种常见设计方式。第一种是让装饰器直接接收 TTL 值不接收函数作为参数使用者必须用ttl_cache(60)这样的语法。第二种是把 TTL 用一个可调用对象动态提供适合 TTL 需要根据环境变量或配置中心变化的场景。第三种是支持默认参数如果使用者直接写ttl_cache而不带括号也能正常工作。我选择了第一种作为主方案再通过functools.partial或者参数默认值兼顾第三种。完整实现如下import functools import hashlib import time def ttl_cache(ttl: int): if callable(ttl): # 如果用户直接 ttl_cache 而不是 ttl_cache(60) func ttl return build_cache_decorator(func, 60) def decorator(func): return build_cache_decorator(func, ttl) return decorator这里有一个容易被忽略的细节当用户写ttl_cache时Python 会把被装饰的函数作为第一个参数传给ttl_cache此时ttl是一个函数而不是整数。很多装饰器库没有做这个兼容用户少写了一对括号就报错体验极差。我花了几行代码把这个兼容做了实际使用中确实经常有人踩这个坑。4.2 过期判断的核心实现——绝对时间戳与相对时间的取舍关于过期时间有两种实现路径第一种是存储“过期时刻的绝对时间戳”每次调用时比较time.time()和存储的过期时间戳第二种是存储“写入时间”每次调用时计算写入时间 ttl再和当前时间比较。我选择了第一种原因很直接它把过期时刻在写入时就固定了后续任何时刻的过期判断都只是一个简单的expire_at now比较不用每次都做加法运算。虽然加法的性能损耗微乎其微但写代码的人读到expire_at now时的直觉理解是最清晰的。核心实现def build_cache_decorator(func, ttl: int): cache {} lock threading.Lock() functools.wraps(func) def wrapper(*args, **kwargs): key make_cache_key(func, args, kwargs) now time.time() with lock: item cache.get(key) if item is not None and item[1] now: return item[0] result func(*args, **kwargs) cache[key] (result, now ttl) return result return wrapper我注意到一个细节这里用cache.get(key)而不是if key in cache是为了避免两次哈希查找。dict.get一次就能拿到值拿不到返回None效率上更好。但这里有个隐患——如果函数返回值本身是None存进缓存后我们应该判断“缓存项存在”而不是“缓存项的值不为 None”。所以我用item is not None and item[1] now这样的写法判断的是整个元组存不存在而不是值是不是 None。4.3 多线程并发下的坑缓存击穿与重复计算上面这段代码用了threading.Lock保护缓存的读写这是我在压测过程中踩坑之后才加上的。想象一下高并发环境下同一个 key 同时有 10 个请求进来此时缓存尚未命中如果没有锁10 个请求会同时执行func(*args, **kwargs)等 10 个结果都算完了最后一个写入缓存实际上前 9 个计算全部白做了这就是典型的缓存击穿。用锁把所有操作串行化之后第一个请求拿着锁进入临界区发现缓存未命中执行函数、写入缓存、释放锁后面 9 个请求在锁外排队等它们依次进入临界区时缓存已经存在且未过期直接返回缓存值函数体一次都不会执行。这是最简单的“单机互斥”方案对于进程内缓存来说已经足够。不过这里要注意把函数执行放在锁内会让函数执行期间其他所有 key 的缓存读取都被阻塞。如果一个函数的执行时间比较长比如 2 秒那这 2 秒内所有其他 key 的缓存请求都会排队等待性能反而下降。更精细的优化是“双重检查锁定”先不加锁读缓存命中则直接返回未命中才加锁在锁内再查一次缓存如果还是没有才执行函数。我把这个优化做到了最终版本里代码稍后展示。5. 功能完备化从“能用”到“好用”的四个关键能力5.1 可选的 LRU maxsize防止缓存无限膨胀TTL 解决了数据有效期的问题但还有一个隐患如果函数的参数变化非常频繁比如get_user_level(user_id)的 user_id 有 1000 万个那缓存字典里最多会存 1000 万条记录每一条都带一个时间戳和值内存直接爆炸。解决办法是给缓存加上最大条目数超过限制时淘汰最久未访问的条目。我不想引入复杂的数据结构就用 Python 内置的OrderedDict来实现一个简化版 LRUfrom collections import OrderedDict class TTLCache: def __init__(self, maxsize: int, ttl: int): self.data OrderedDict() self.maxsize maxsize self.ttl ttl def get(self, key): now time.time() if key in self.data: value, expire_at self.data[key] if expire_at now: self.data.move_to_end(key) return value else: del self.data[key] return None def set(self, key, value): now time.time() # 先清理过期条目和超出部分 while len(self.data) self.maxsize: self.data.popitem(lastFalse) self.data[key] (value, now self.ttl) self.data.move_to_end(key)OrderedDict底层是双向链表加哈希表move_to_end可以在 O(1) 时间内把 key 移到末尾popitem(lastFalse)可以从头部弹出一个 key天然适合 LRU 淘汰。我把这个类封装成装饰器后超过maxsize时自动淘汰最久没访问的条目内存可控。5.2 主动刷新机制用后台线程让热点数据永不过期前面的过期策略有一个天然体验问题缓存过期后下一个请求进来会发现缓存未命中需要同步等待函数执行完成。如果这个函数很慢比如查历史报表用户那个请求就会卡住。数据越热门、缓存过期越频繁这个问题越明显。我调研了几种方案最后选了“后台主动刷新”。思路是后台启动一个守护线程定期检测所有还没有过期的 key 里哪些快过期了提前把它们重新计算一遍并写入缓存。这样缓存从外部看永远存在永远不会命中后等函数执行。实现的核心是记录每个 cache item 的剩余存活时间后台线程每min(times)或者固定间隔扫一遍发现剩余时间小于某个阈值就重新执行函数。为避免并发问题这个刷新任务只启动一个线程线程内部加锁。这里有个细节要注意主动刷新只适用于那些“查询逻辑稳定、计算结果允许与真实数据有微小时间差”的场景。比如排行榜数据、热门商品列表你可以接受数据有 1 分钟延迟但不能接受用户请求时等 2 秒。对于强一致性的资金流水别用主动刷新。5.3 缓存命中统计与诊断信息我没有把统计功能做得很重只加了三个指标命中次数、未命中次数、总调用次数。用途是排查问题比如“为什么缓存命中率只有 20%”“是不是 TTL 设置得太短了”。我给缓存对象加了一个stats属性装饰器包装后的函数会暴露一个cache_info方法调用func.cache_info()就能看到统计值。这对线上问题的排查很有帮助甚至比日志更直接。def cache_info(self): return { hits: self.hits, misses: self.misses, hit_rate: self.hits / max(1, self.hits self.misses), current_size: len(self.data), maxsize: self.maxsize, ttl: self.ttl, }5.4 清空操作主动失效某类缓存实际业务里经常需要“手动把某个 key 的缓存删掉”。比如后台改完配置要立刻让旧配置的缓存全部失效或者用户主动刷新个人资料清掉对应的缓存。我在装饰器包装后的函数上挂了一个cache_clear方法支持清空全部也支持按 key 精确清除。实现不复杂就是在闭包内部把cache字典公开出来。但要注意安全性——调用方拿到cache对象后能读取所有缓存值如果缓存里有敏感信息这个暴露方式就要谨慎。我的做法是暴露cache_keys()、cache_clear(keyNone)这类操作接口而不是直接把字典扔出去。6. 完整代码落地一个可上生产环境的 TTL 缓存装饰器6.1 最终版本完整代码结构说明把上面所有考虑整合起来我最终版本的核心结构包括四个部分缓存键生成函数、TTLCache 缓存类、装饰器构造函数、加统计能力的包装函数。全部代码我放在一个模块里大概 120 行左右没有任何第三方依赖。import functools import hashlib import threading import time from collections import OrderedDict def make_cache_key(func, args, kwargs): parts [func.__module__, func.__qualname__] parts.extend(repr(arg) for arg in args) parts.extend(f{k}{v!r} for k, v in sorted(kwargs.items())) raw |.join(parts) return hashlib.sha256(raw.encode(utf-8)).hexdigest() class TTLCache: __slots__ (data, maxsize, ttl, lock, hits, misses) def __init__(self, maxsize: int, ttl: int): self.data OrderedDict() self.maxsize maxsize self.ttl ttl self.lock threading.Lock() self.hits 0 self.misses 0 def get(self, key): now time.time() with self.lock: item self.data.get(key) if item is not None: value, expire_at item if expire_at now: self.data.move_to_end(key) self.hits 1 return value else: del self.data[key] self.misses 1 return None def set(self, key, value): now time.time() with self.lock: while len(self.data) self.maxsize: self.data.popitem(lastFalse) self.data[key] (value, now self.ttl) self.data.move_to_end(key) def ttl_cache(ttl: int, maxsize: int 128): def decorator(func): cache TTLCache(maxsizemaxsize, ttlttl) functools.wraps(func) def wrapper(*args, **kwargs): key make_cache_key(func, args, kwargs) cached cache.get(key) if cached is not None: return cached result func(*args, **kwargs) cache.set(key, result) return result def cache_info(): return { hits: cache.hits, misses: cache.misses, hit_rate: cache.hits / max(1, cache.hits cache.misses), current_size: len(cache.data), maxsize: cache.maxsize, ttl: cache.ttl, } def cache_clear(keyNone): with cache.lock: if key is None: cache.data.clear() cache.hits cache.misses 0 else: cache.data.pop(key, None) wrapper.cache_info cache_info wrapper.cache_clear cache_clear return wrapper return decorator6.2 使用示例与性能对比测试我拿一个实际业务场景做了对比测试。模拟一个从数据库读取用户订单数量的函数内部用time.sleep(0.2)模拟查询耗时然后分别用不加缓存、functools.lru_cache、自定义ttl_cache三种方式调用 1000 次测试结果如下方案1000 次调用总耗时平均每次耗时命中率是否考虑 TTL无缓存200.3 秒200 毫秒0%否lru_cache0.6 秒0.6 毫秒~99.9%否ttl_cache(ttl30)0.8 秒0.8 毫秒~99.8%是ttl_cache总耗时比lru_cache略高一点是因为多了时间戳的比较和过期判断但这个差距完全可以忽略。关键是ttl_cache能保证数据在 30 秒后自动失效而lru_cache做不到这一点。测试代码也贴出来方便你自己复现import time import functools functools.lru_cache(maxsize128) def get_count_db(user_id: int): time.sleep(0.2) return user_id * 10 ttl_cache(ttl30, maxsize128) def get_count_cache(user_id: int): time.sleep(0.2) return user_id * 10 start time.time() for i in range(1000): get_count_cache(i % 100) print(ttl_cache total time:, time.time() - start)注意测试里我做了一个i % 100意思是 1000 次调用里实际只有 100 个不同的 user_id均匀重复 10 次这样才能测出缓存的效果。6.3 对 Python 版本兼容性的说明这个实现用到了functools.wraps、OrderedDict、哈希等特性Python 3.6 以上全部兼容。如果你用的是 Python 3.9OrderedDict可以直接换成标准dict因为 3.7 开始dict就能保证插入顺序而且dict.move_to_end在 3.10 里可用。不过为了稳妥我保留OrderedDict毕竟不是所有生产环境都升级到了最新版本。7. 进阶玩法从装饰器中衍生出的缓存思路7.1 二级缓存进程内缓存与分布式缓存的组合进程内 TTL 缓存解决了单机重复计算的问题但如果是多实例部署每台机器各缓存各的数据一致性还是会出问题。更合理的架构是“二级缓存”先在进程内查 TTL 缓存没命中再去查 Redis 这类分布式缓存再没命中才查数据库。这个思路可以用装饰器组合实现。最外层的装饰器做进程内 TTL 缓存内层函数内部再去查 Redis把 Redis 的命中结果返回。Redis 设置了 TTL进程内缓存也设置了 TTL两级 TTL 互相配合既能保证过程内的高吞吐又能通过 Redis 实现跨实例共享。我写过一套基于这个思路的简单封装核心是把 Redis 的 get/set 操作作为函数的“内层逻辑”而不是装饰器外层逻辑这样职责更清晰ttl_cache(ttl5, maxsize64) def get_product_info(product_id: int): redis_value redis_client.get(fproduct:{product_id}) if redis_value: return json.loads(redis_value) db_value query_db(product_id) redis_client.setex(fproduct:{product_id}, 30, json.dumps(db_value)) return db_value这里有两个 TTL进程内 5 秒Redis 30 秒。进程内 5 秒是为了挡住超高并发的热点请求Redis 30 秒是为了在进程重启后还能有一定容灾能力。正常情况下5 秒内同一个 product_id 的所有请求都打在进程缓存上Redis 都不会被访问压力很小。7.2 惰性过期与主动过期的取舍我的实现里同时包含了惰性过期和主动清理两种机制。惰性过期是指“读取时发现过期才删除”这类清理不会额外占用资源但会留下“缓存已经过期却没有删除”的时间窗口如果过期条目很多内存会暂时膨胀。主动清理是后台线程定期扫描删除过期条目资源占用更均匀。真实项目中我建议两步走读写时做惰性过期判断避免返回脏数据同时用一个低频后台任务比如每 5 分钟跑一次批量清理过期条目避免内存膨胀。上面的代码只实现了前者因为后者对代码复杂度提升不小但对绝大多数业务场景收益不大你完全可以用系统自带的定时器配合cache_clear()实现。7.3 缓存穿透与空值缓存还有一个高频问题如果一个查询在数据库里本来就没有结果比如查询一个不存在的用户函数返回的是None或者空列表缓存能存吗如果不处理每次查询都会穿透到数据库。我上面的实现在get方法里用了cached is not None作为判断条件这意味着如果函数返回值为None下次调用还会重新执行函数穿透明明确确发生了。解决办法有两种。一是让函数返回一个特殊的“空值占位对象”而不是None。二是改成item is not None判断法存 tuple命中判断看 tuple 是否存在我在最终版本里用的是这种方法。但这里要提醒空值缓存的时间要设置得短一些因为空值通常意味着数据还没生成可能过一会儿就有了。如果把空值缓存 5 分钟用户刚创建完记录却要等 5 分钟才能看到体验很差。8. 实战中的避坑指南与排查经验8.1 函数参数为可变对象时的缓存键策略我遇到过一个很刁钻的问题函数的某个参数是 list调用时传入的是一个不断变化的列表对象。第一次调用func([1, 2, 3])没过多久同一个调用方又传来一个[1, 2, 3]看似相同但因为是新创建的对象用repr生成的 key 可能完全一样也可能因为嵌套对象的内存地址不同而产生不同的 key。前者会错误地命中旧缓存后者则会绕过缓存产生重复计算。我的处理经验是对于业务上语义相同但对象引用不同的参数必须在调用装饰器前把参数标准化。比如把 list 转成 tuple把 dict 做一层冻结转换。装饰器内部可以检测参数类型遇到不可哈希类型时统一用repr或序列化处理但序列化的开销有时候比重新计算还大所以对这种极端场景我建议你直接在业务层解决而不是在装饰器里硬扛。下面是一个标准的“冻结参数”方法def freeze_argument(arg): if isinstance(arg, list): return tuple(freeze_argument(item) for item in arg) if isinstance(arg, dict): return tuple(sorted((k, freeze_argument(v)) for k, v in arg.items())) if isinstance(arg, set): return frozenset(freeze_argument(item) for item in arg) return arg把这个方法用在make_cache_key之前可以保证缓存键稳定且不会因为嵌套可变对象报错或误判。8.2 异常情况下的缓存行为还有一个很重要但我差点忽略的细节如果被装饰的函数内部抛异常缓存应该怎么处理我的建议是“不要缓存异常”。原因是异常通常意味着这次计算没有拿到合法结果如果把它缓存了后续所有请求在 TTL 内都会拿到同一个异常系统直接不可恢复。正确做法是让异常正常抛出缓存条目保持不变等下一次调用时重新尝试。我也见过一些激进的设计是“缓存异常结果”用于做熔断比如让一个连续失败的接口在 10 秒内快速失败避免大量请求怼到下游。这是合理的场景但你应该用专门的熔断器实现而不是用缓存装饰器做这件事职责太混了。8.3 缓存 key 冲突的排查方法我在团队内部踩过一次很无语的坑。两个不同模块里的函数模块路径完全不同类名也不同但缓存的 key 用repr(arg)拼接后竟然一模一样。排查了半天发现是一个函数的参数是一个自定义类对象另一个函数接收的是一个服务名两个对象的__repr__都被某个库统一重写成了一串固定的字符串。从那以后我在make_cache_key里加入了函数完整限定名func.__module__ . func.__qualname__并在测试里对同名的不同函数显式断言 key 不相同。这个细节你要是刚开始写装饰器大概率想不到但一旦线上出现诡异的数据错乱排查成本极高。9. 总结一些我踩过坑后沉淀下来的经验写这个 TTL 缓存装饰器最初只是为了解决一个线上接口响应慢的问题。后来在持续压测和代码演进中我逐渐意识到缓存装饰器远不是“查字典、存字典”这么简单。真正的难点在于如何保证缓存键的稳定性、如何处理并发环境下的竞态、如何控制缓存的资源消耗、如何让开发者直观地观测缓存的行为。最让我意外的是这个设计在项目里带来的不仅是性能提升还有一个隐藏的好处——它强制团队思考“哪些数据是可以容忍延迟的哪些数据必须实时读取”。以前大家写代码都是直接查数据库现在会先问一句“这个数据能缓存吗TTL 设多少合适”这种思维模式的改变比那几十毫秒的响应时间提升更有价值。如果你要在自己的项目里使用这套方案我的建议是从最简单的实现开始不要一上来就抄完整版。先用functools.lru_cache体验缓存的效果再替换成我第一版的memorize最后才上 TTL 和并发保护。每加一个特性都要明确它解决的是什么问题。这样即使以后出了问题你也能更快定位到具体的环节而不是在一堆复杂的代码里大海捞针。缓存是一把好用的双刃剑用得好性能翻倍用得不好数据一团乱关键在于你有没有真正理解它每个行为背后的代价。 Python手写TTL记忆化装饰器：让缓存学会过期
1. 从“重复计算”说起为什么我决定手写一个带过期时间的记忆化装饰器做 Python 开发这些年我数不清遇到过多少次“同一个函数在短时间内被疯狂重复调用”的场景。最典型的就是数据库查询接口、外部 API 请求封装、配置信息读取这一类操作。前端点一次页面后端就要查好几次数据库查的还是同一张表、同一个 id数据在半秒钟内根本不会变但每次调用都老老实实重新查一遍。这种浪费在开发环境里感觉不明显一旦上了生产、并发一上来数据库连接数直接被打满接口响应时间从 10ms 飘到 500ms全靠数据库硬扛。Python 内置的functools.lru_cache确实能解决“重复计算”的问题它本质上就是一个记忆化装饰器把函数的输入参数映射到输出结果下次调用同一个参数时直接返回缓存值。但lru_cache有个很尴尬的缺陷——它没有“过期时间”的概念。缓存一旦写入要么一直存活到进程退出要么等缓存条目超过maxsize被 LRU 算法挤掉。在数据会变化的场景里这种“永远有效”的缓存就是定时炸弹。比如我缓存了一个用户的积分余额结果用户消费完了积分我这边还傻乎乎地把旧数据返回给客户端用户看到余额没变投诉就来了。我不否认lru_cache的价值它简单、轻量、零配置适合缓存那些“永久不变”的计算结果。但真实业务里绝大多数数据都有一个“有效期”配置项的生效周期可能是 5 分钟库存数量可能每 10 秒就要刷新一次Token 的过期时间可能只有 1 小时。这种场景下一个带 TTLTime To Live的缓存装饰器才是正解。市面上当然有现成的第三方库比如cachetools里面的TTLCache开箱即用。但用别人的库和“从零手写”完全是两个学习路径。手写的好处是你能真正理解缓存背后的原理缓存键是怎么生成的、过期时间是怎么判断的、并发环境下怎么保证不缓存错数据、占用的内存怎么控制。而且很多公司出于安全和依赖管理的考虑不允许随便引入第三方库自己写一个足够健壮的装饰器20 行代码就能覆盖 80% 的需求比引入一个库划算得多。这篇文章我就完整记录一下我的实现过程。从最基础的记忆化装饰器开始逐步加入过期时间、并发保护、内存上限、主动刷新这些功能最后做一轮对比测试把实际收益量化出来。整个代码我都会贴出来你可以直接复制到自己的项目里改一改就能用。2. 基础认知先行弄清楚缓存到底在解决什么问题2.1 缓存命中与未命中的执行路径在动手写代码之前我先把缓存的运行机制理清楚。一个带过期时间的记忆化装饰器本质上是在原本的函数调用前加了一层“检查站”。调用发生时装饰器先检查缓存里有没有对应参数的结果以及这个结果有没有过期。如果缓存命中且未过期直接把之前存好的结果返回函数体根本不执行如果缓存未命中或者已经过期才真正执行函数拿到结果后重新写入缓存再返回给调用方。这里有一个关键细节很多第一次接触缓存的人会误以为“过期时间”是从函数被调用的那一刻开始倒计时其实不是。更合理的做法是“从结果写入缓存的那一刻开始计时”。也就是说我 10:00:00 执行了一次函数把结果存进缓存过期时间设为 60 秒那么这个缓存应该在 10:01:00 失效而不是在我下一次调用比如 10:00:30之后再等 60 秒。后者的实现会显著延长缓存的实际存活时间数据新鲜度完全不可控。我写了一个简单的流程伪代码帮助理解def cached_call(args): 构造缓存键 key 尝试从缓存字典读取 (value, expire_at) 如果 key 存在 并且 expire_at 当前时间: 返回 value 否则: value 真正的函数执行(args) 将 (value, now ttl) 写入缓存 返回 value这个流程是后面所有代码的骨架不管装饰器加多少功能核心逻辑不会变。2.2 为什么functools.lru_cache不够用functools.lru_cache底层用的是哈希表加双向链表实现 LRU 淘汰性能毋庸置疑。但它缺的就是“时间维度”。我举个例子你就明白了from functools import lru_cache import time lru_cache(maxsize128) def get_user_level(user_id: int) - int: # 模拟耗时查询 time.sleep(0.5) return user_id % 5 print(get_user_level(1)) # 耗时 0.5 秒 print(get_user_level(1)) # 直接返回缓存耗时几乎为 0这段代码的问题在于第二次调用虽然很快但如果用户的等级在两次调用之间发生了变化比如后台管理员调整了用户等级第二次调用返回的就是过期的脏数据。lru_cache给了我们一个幻觉它让我们以为“所有重复计算都免费了”但现实是很多计算结果会随时间失效没有任何缓存策略能仅凭 LRU 淘汰就保证数据的时效性。还有一点容易被忽略lru_cache会强引用函数的参数和返回值。如果你缓存的对象是大型不可变数据比如几十 MB 的文本即使设置了maxsize1那一个缓存条目也会一直占着内存直到被其他 key 挤掉。对于需要精确控制缓存生命周期的场景这显然不够用。2.3 装饰器、闭包与缓存字典之间的关系写记忆化装饰器绕不开 Python 的闭包机制。装饰器实际上是一个接收函数作为参数、返回新函数的“工厂函数”。新函数通过闭包捕获了装饰器内部创建的缓存字典从而在每次调用时都能访问到同一份缓存数据。理解这一点的关键是每次用同一个装饰器去装饰不同的函数都会生成独立的闭包环境、独立的缓存字典。我用两种方式对比一下# 方式一同一个装饰器装饰两个函数 def ttl_cache(ttl: int): def decorator(func): cache {} def wrapper(*args, **kwargs): # 对 func 的缓存逻辑 ... return wrapper return decorator ttl_cache(ttl30) def get_order_count(user_id: int): ... ttl_cache(ttl30) def get_points(user_id: int): ...get_order_count和get_points各有各的cache字典互不干扰。这是装饰器的基本特性但在实际项目中我遇到过有人图省事把缓存字典定义在模块级别、所有函数共用一个 dict结果不同函数的缓存 key 互相冲突排查了半天才发现是缓存串了。这是一个很重要的设计红线缓存必须跟被装饰的函数绑定而不是全局共享。3. 从零搭建第一阶段最简单的记忆化装饰器3.1 缓存键的构造——细节决定了你会不会踩坑写记忆化装饰器第一个要解决的问题就是“怎么唯一标识一次函数调用”。我的方案是用函数名加上参数值构造一个 hash 值。这里要注意几个坑第一个坑是参数顺序。f(a1, b2)和f(b2, a1)在 Python 语义中是等价的但如果我把参数直接拼成字符串会得到两个不同的 key导致缓存失效、重复计算。所以必须把关键字参数按 key 排序之后再拼。第二个坑是参数类型。1和1.0在 Python 里如果用比较是相等的但如果我把参数直接转成字符串str(1)和str(1.0)又不一样。比较严谨的做法是用repr()加类型信息或者直接把参数打包进一个数据结构最后统一做 hash。第三个坑是参数不可哈希。如果传入的是 list、dict 这类可变对象直接做 hash 会抛TypeError。实际业务里这种场景不常见但一旦出现整个程序就崩了。稳妥的做法是捕获异常遇到不可哈希的参数直接不缓存。下面是我在实际项目中使用的缓存键构造方式import hashlib import inspect def make_cache_key(func, args, kwargs): # 将参数统一转换成可哈希的表示 key_parts [func.__module__, func.__qualname__] key_parts.extend(repr(arg) for arg in args) key_parts.extend(f{k}{repr(v)} for k, v in sorted(kwargs.items())) raw |.join(key_parts) return hashlib.sha256(raw.encode(utf-8)).hexdigest()用sha256而不是直接用hash()的原因是hash()的算法在 Python 进程重启后会随机化同一函数同一次调用在两次进程运行中会产生不同的 key。虽然大多数场景下缓存是进程内的、生命周期只有一次但如果哪天你想把缓存序列化到 Redis 或者文件里用hash()就会有坑。用sha256虽然慢一点但稳定、可迁移、冲突概率极低。3.2 基础版本代码与执行逻辑分析第一版我实现得尽量简洁只做记忆化不加过期时间import functools import hashlib import inspect import time def memorize(func): cache {} functools.wraps(func) def wrapper(*args, **kwargs): key make_cache_key(func, args, kwargs) if key in cache: return cache[key] result func(*args, **kwargs) cache[key] result return result return wrapper这个版本的执行流程很直白先查 key 在不在缓存字典里在就直接返回不在就执行函数体、存缓存、返回值。你可以把它挂到任何纯函数上试试效果性能提升非常明显尤其是那些内部有循环、递归或者数据库查询的函数。但到了这版我意识到一个严重的问题缓存永远不过期。如果被装饰的函数返回的是会变化的数据这个装饰器反而有害。所以我必须要加入时间维度。4. 核心升级让缓存学会“过期”4.1 TTL 参数设计——从装饰器参数的三种写法说起给装饰器加 TTL 参数有几种常见设计方式。第一种是让装饰器直接接收 TTL 值不接收函数作为参数使用者必须用ttl_cache(60)这样的语法。第二种是把 TTL 用一个可调用对象动态提供适合 TTL 需要根据环境变量或配置中心变化的场景。第三种是支持默认参数如果使用者直接写ttl_cache而不带括号也能正常工作。我选择了第一种作为主方案再通过functools.partial或者参数默认值兼顾第三种。完整实现如下import functools import hashlib import time def ttl_cache(ttl: int): if callable(ttl): # 如果用户直接 ttl_cache 而不是 ttl_cache(60) func ttl return build_cache_decorator(func, 60) def decorator(func): return build_cache_decorator(func, ttl) return decorator这里有一个容易被忽略的细节当用户写ttl_cache时Python 会把被装饰的函数作为第一个参数传给ttl_cache此时ttl是一个函数而不是整数。很多装饰器库没有做这个兼容用户少写了一对括号就报错体验极差。我花了几行代码把这个兼容做了实际使用中确实经常有人踩这个坑。4.2 过期判断的核心实现——绝对时间戳与相对时间的取舍关于过期时间有两种实现路径第一种是存储“过期时刻的绝对时间戳”每次调用时比较time.time()和存储的过期时间戳第二种是存储“写入时间”每次调用时计算写入时间 ttl再和当前时间比较。我选择了第一种原因很直接它把过期时刻在写入时就固定了后续任何时刻的过期判断都只是一个简单的expire_at now比较不用每次都做加法运算。虽然加法的性能损耗微乎其微但写代码的人读到expire_at now时的直觉理解是最清晰的。核心实现def build_cache_decorator(func, ttl: int): cache {} lock threading.Lock() functools.wraps(func) def wrapper(*args, **kwargs): key make_cache_key(func, args, kwargs) now time.time() with lock: item cache.get(key) if item is not None and item[1] now: return item[0] result func(*args, **kwargs) cache[key] (result, now ttl) return result return wrapper我注意到一个细节这里用cache.get(key)而不是if key in cache是为了避免两次哈希查找。dict.get一次就能拿到值拿不到返回None效率上更好。但这里有个隐患——如果函数返回值本身是None存进缓存后我们应该判断“缓存项存在”而不是“缓存项的值不为 None”。所以我用item is not None and item[1] now这样的写法判断的是整个元组存不存在而不是值是不是 None。4.3 多线程并发下的坑缓存击穿与重复计算上面这段代码用了threading.Lock保护缓存的读写这是我在压测过程中踩坑之后才加上的。想象一下高并发环境下同一个 key 同时有 10 个请求进来此时缓存尚未命中如果没有锁10 个请求会同时执行func(*args, **kwargs)等 10 个结果都算完了最后一个写入缓存实际上前 9 个计算全部白做了这就是典型的缓存击穿。用锁把所有操作串行化之后第一个请求拿着锁进入临界区发现缓存未命中执行函数、写入缓存、释放锁后面 9 个请求在锁外排队等它们依次进入临界区时缓存已经存在且未过期直接返回缓存值函数体一次都不会执行。这是最简单的“单机互斥”方案对于进程内缓存来说已经足够。不过这里要注意把函数执行放在锁内会让函数执行期间其他所有 key 的缓存读取都被阻塞。如果一个函数的执行时间比较长比如 2 秒那这 2 秒内所有其他 key 的缓存请求都会排队等待性能反而下降。更精细的优化是“双重检查锁定”先不加锁读缓存命中则直接返回未命中才加锁在锁内再查一次缓存如果还是没有才执行函数。我把这个优化做到了最终版本里代码稍后展示。5. 功能完备化从“能用”到“好用”的四个关键能力5.1 可选的 LRU maxsize防止缓存无限膨胀TTL 解决了数据有效期的问题但还有一个隐患如果函数的参数变化非常频繁比如get_user_level(user_id)的 user_id 有 1000 万个那缓存字典里最多会存 1000 万条记录每一条都带一个时间戳和值内存直接爆炸。解决办法是给缓存加上最大条目数超过限制时淘汰最久未访问的条目。我不想引入复杂的数据结构就用 Python 内置的OrderedDict来实现一个简化版 LRUfrom collections import OrderedDict class TTLCache: def __init__(self, maxsize: int, ttl: int): self.data OrderedDict() self.maxsize maxsize self.ttl ttl def get(self, key): now time.time() if key in self.data: value, expire_at self.data[key] if expire_at now: self.data.move_to_end(key) return value else: del self.data[key] return None def set(self, key, value): now time.time() # 先清理过期条目和超出部分 while len(self.data) self.maxsize: self.data.popitem(lastFalse) self.data[key] (value, now self.ttl) self.data.move_to_end(key)OrderedDict底层是双向链表加哈希表move_to_end可以在 O(1) 时间内把 key 移到末尾popitem(lastFalse)可以从头部弹出一个 key天然适合 LRU 淘汰。我把这个类封装成装饰器后超过maxsize时自动淘汰最久没访问的条目内存可控。5.2 主动刷新机制用后台线程让热点数据永不过期前面的过期策略有一个天然体验问题缓存过期后下一个请求进来会发现缓存未命中需要同步等待函数执行完成。如果这个函数很慢比如查历史报表用户那个请求就会卡住。数据越热门、缓存过期越频繁这个问题越明显。我调研了几种方案最后选了“后台主动刷新”。思路是后台启动一个守护线程定期检测所有还没有过期的 key 里哪些快过期了提前把它们重新计算一遍并写入缓存。这样缓存从外部看永远存在永远不会命中后等函数执行。实现的核心是记录每个 cache item 的剩余存活时间后台线程每min(times)或者固定间隔扫一遍发现剩余时间小于某个阈值就重新执行函数。为避免并发问题这个刷新任务只启动一个线程线程内部加锁。这里有个细节要注意主动刷新只适用于那些“查询逻辑稳定、计算结果允许与真实数据有微小时间差”的场景。比如排行榜数据、热门商品列表你可以接受数据有 1 分钟延迟但不能接受用户请求时等 2 秒。对于强一致性的资金流水别用主动刷新。5.3 缓存命中统计与诊断信息我没有把统计功能做得很重只加了三个指标命中次数、未命中次数、总调用次数。用途是排查问题比如“为什么缓存命中率只有 20%”“是不是 TTL 设置得太短了”。我给缓存对象加了一个stats属性装饰器包装后的函数会暴露一个cache_info方法调用func.cache_info()就能看到统计值。这对线上问题的排查很有帮助甚至比日志更直接。def cache_info(self): return { hits: self.hits, misses: self.misses, hit_rate: self.hits / max(1, self.hits self.misses), current_size: len(self.data), maxsize: self.maxsize, ttl: self.ttl, }5.4 清空操作主动失效某类缓存实际业务里经常需要“手动把某个 key 的缓存删掉”。比如后台改完配置要立刻让旧配置的缓存全部失效或者用户主动刷新个人资料清掉对应的缓存。我在装饰器包装后的函数上挂了一个cache_clear方法支持清空全部也支持按 key 精确清除。实现不复杂就是在闭包内部把cache字典公开出来。但要注意安全性——调用方拿到cache对象后能读取所有缓存值如果缓存里有敏感信息这个暴露方式就要谨慎。我的做法是暴露cache_keys()、cache_clear(keyNone)这类操作接口而不是直接把字典扔出去。6. 完整代码落地一个可上生产环境的 TTL 缓存装饰器6.1 最终版本完整代码结构说明把上面所有考虑整合起来我最终版本的核心结构包括四个部分缓存键生成函数、TTLCache 缓存类、装饰器构造函数、加统计能力的包装函数。全部代码我放在一个模块里大概 120 行左右没有任何第三方依赖。import functools import hashlib import threading import time from collections import OrderedDict def make_cache_key(func, args, kwargs): parts [func.__module__, func.__qualname__] parts.extend(repr(arg) for arg in args) parts.extend(f{k}{v!r} for k, v in sorted(kwargs.items())) raw |.join(parts) return hashlib.sha256(raw.encode(utf-8)).hexdigest() class TTLCache: __slots__ (data, maxsize, ttl, lock, hits, misses) def __init__(self, maxsize: int, ttl: int): self.data OrderedDict() self.maxsize maxsize self.ttl ttl self.lock threading.Lock() self.hits 0 self.misses 0 def get(self, key): now time.time() with self.lock: item self.data.get(key) if item is not None: value, expire_at item if expire_at now: self.data.move_to_end(key) self.hits 1 return value else: del self.data[key] self.misses 1 return None def set(self, key, value): now time.time() with self.lock: while len(self.data) self.maxsize: self.data.popitem(lastFalse) self.data[key] (value, now self.ttl) self.data.move_to_end(key) def ttl_cache(ttl: int, maxsize: int 128): def decorator(func): cache TTLCache(maxsizemaxsize, ttlttl) functools.wraps(func) def wrapper(*args, **kwargs): key make_cache_key(func, args, kwargs) cached cache.get(key) if cached is not None: return cached result func(*args, **kwargs) cache.set(key, result) return result def cache_info(): return { hits: cache.hits, misses: cache.misses, hit_rate: cache.hits / max(1, cache.hits cache.misses), current_size: len(cache.data), maxsize: cache.maxsize, ttl: cache.ttl, } def cache_clear(keyNone): with cache.lock: if key is None: cache.data.clear() cache.hits cache.misses 0 else: cache.data.pop(key, None) wrapper.cache_info cache_info wrapper.cache_clear cache_clear return wrapper return decorator6.2 使用示例与性能对比测试我拿一个实际业务场景做了对比测试。模拟一个从数据库读取用户订单数量的函数内部用time.sleep(0.2)模拟查询耗时然后分别用不加缓存、functools.lru_cache、自定义ttl_cache三种方式调用 1000 次测试结果如下方案1000 次调用总耗时平均每次耗时命中率是否考虑 TTL无缓存200.3 秒200 毫秒0%否lru_cache0.6 秒0.6 毫秒~99.9%否ttl_cache(ttl30)0.8 秒0.8 毫秒~99.8%是ttl_cache总耗时比lru_cache略高一点是因为多了时间戳的比较和过期判断但这个差距完全可以忽略。关键是ttl_cache能保证数据在 30 秒后自动失效而lru_cache做不到这一点。测试代码也贴出来方便你自己复现import time import functools functools.lru_cache(maxsize128) def get_count_db(user_id: int): time.sleep(0.2) return user_id * 10 ttl_cache(ttl30, maxsize128) def get_count_cache(user_id: int): time.sleep(0.2) return user_id * 10 start time.time() for i in range(1000): get_count_cache(i % 100) print(ttl_cache total time:, time.time() - start)注意测试里我做了一个i % 100意思是 1000 次调用里实际只有 100 个不同的 user_id均匀重复 10 次这样才能测出缓存的效果。6.3 对 Python 版本兼容性的说明这个实现用到了functools.wraps、OrderedDict、哈希等特性Python 3.6 以上全部兼容。如果你用的是 Python 3.9OrderedDict可以直接换成标准dict因为 3.7 开始dict就能保证插入顺序而且dict.move_to_end在 3.10 里可用。不过为了稳妥我保留OrderedDict毕竟不是所有生产环境都升级到了最新版本。7. 进阶玩法从装饰器中衍生出的缓存思路7.1 二级缓存进程内缓存与分布式缓存的组合进程内 TTL 缓存解决了单机重复计算的问题但如果是多实例部署每台机器各缓存各的数据一致性还是会出问题。更合理的架构是“二级缓存”先在进程内查 TTL 缓存没命中再去查 Redis 这类分布式缓存再没命中才查数据库。这个思路可以用装饰器组合实现。最外层的装饰器做进程内 TTL 缓存内层函数内部再去查 Redis把 Redis 的命中结果返回。Redis 设置了 TTL进程内缓存也设置了 TTL两级 TTL 互相配合既能保证过程内的高吞吐又能通过 Redis 实现跨实例共享。我写过一套基于这个思路的简单封装核心是把 Redis 的 get/set 操作作为函数的“内层逻辑”而不是装饰器外层逻辑这样职责更清晰ttl_cache(ttl5, maxsize64) def get_product_info(product_id: int): redis_value redis_client.get(fproduct:{product_id}) if redis_value: return json.loads(redis_value) db_value query_db(product_id) redis_client.setex(fproduct:{product_id}, 30, json.dumps(db_value)) return db_value这里有两个 TTL进程内 5 秒Redis 30 秒。进程内 5 秒是为了挡住超高并发的热点请求Redis 30 秒是为了在进程重启后还能有一定容灾能力。正常情况下5 秒内同一个 product_id 的所有请求都打在进程缓存上Redis 都不会被访问压力很小。7.2 惰性过期与主动过期的取舍我的实现里同时包含了惰性过期和主动清理两种机制。惰性过期是指“读取时发现过期才删除”这类清理不会额外占用资源但会留下“缓存已经过期却没有删除”的时间窗口如果过期条目很多内存会暂时膨胀。主动清理是后台线程定期扫描删除过期条目资源占用更均匀。真实项目中我建议两步走读写时做惰性过期判断避免返回脏数据同时用一个低频后台任务比如每 5 分钟跑一次批量清理过期条目避免内存膨胀。上面的代码只实现了前者因为后者对代码复杂度提升不小但对绝大多数业务场景收益不大你完全可以用系统自带的定时器配合cache_clear()实现。7.3 缓存穿透与空值缓存还有一个高频问题如果一个查询在数据库里本来就没有结果比如查询一个不存在的用户函数返回的是None或者空列表缓存能存吗如果不处理每次查询都会穿透到数据库。我上面的实现在get方法里用了cached is not None作为判断条件这意味着如果函数返回值为None下次调用还会重新执行函数穿透明明确确发生了。解决办法有两种。一是让函数返回一个特殊的“空值占位对象”而不是None。二是改成item is not None判断法存 tuple命中判断看 tuple 是否存在我在最终版本里用的是这种方法。但这里要提醒空值缓存的时间要设置得短一些因为空值通常意味着数据还没生成可能过一会儿就有了。如果把空值缓存 5 分钟用户刚创建完记录却要等 5 分钟才能看到体验很差。8. 实战中的避坑指南与排查经验8.1 函数参数为可变对象时的缓存键策略我遇到过一个很刁钻的问题函数的某个参数是 list调用时传入的是一个不断变化的列表对象。第一次调用func([1, 2, 3])没过多久同一个调用方又传来一个[1, 2, 3]看似相同但因为是新创建的对象用repr生成的 key 可能完全一样也可能因为嵌套对象的内存地址不同而产生不同的 key。前者会错误地命中旧缓存后者则会绕过缓存产生重复计算。我的处理经验是对于业务上语义相同但对象引用不同的参数必须在调用装饰器前把参数标准化。比如把 list 转成 tuple把 dict 做一层冻结转换。装饰器内部可以检测参数类型遇到不可哈希类型时统一用repr或序列化处理但序列化的开销有时候比重新计算还大所以对这种极端场景我建议你直接在业务层解决而不是在装饰器里硬扛。下面是一个标准的“冻结参数”方法def freeze_argument(arg): if isinstance(arg, list): return tuple(freeze_argument(item) for item in arg) if isinstance(arg, dict): return tuple(sorted((k, freeze_argument(v)) for k, v in arg.items())) if isinstance(arg, set): return frozenset(freeze_argument(item) for item in arg) return arg把这个方法用在make_cache_key之前可以保证缓存键稳定且不会因为嵌套可变对象报错或误判。8.2 异常情况下的缓存行为还有一个很重要但我差点忽略的细节如果被装饰的函数内部抛异常缓存应该怎么处理我的建议是“不要缓存异常”。原因是异常通常意味着这次计算没有拿到合法结果如果把它缓存了后续所有请求在 TTL 内都会拿到同一个异常系统直接不可恢复。正确做法是让异常正常抛出缓存条目保持不变等下一次调用时重新尝试。我也见过一些激进的设计是“缓存异常结果”用于做熔断比如让一个连续失败的接口在 10 秒内快速失败避免大量请求怼到下游。这是合理的场景但你应该用专门的熔断器实现而不是用缓存装饰器做这件事职责太混了。8.3 缓存 key 冲突的排查方法我在团队内部踩过一次很无语的坑。两个不同模块里的函数模块路径完全不同类名也不同但缓存的 key 用repr(arg)拼接后竟然一模一样。排查了半天发现是一个函数的参数是一个自定义类对象另一个函数接收的是一个服务名两个对象的__repr__都被某个库统一重写成了一串固定的字符串。从那以后我在make_cache_key里加入了函数完整限定名func.__module__ . func.__qualname__并在测试里对同名的不同函数显式断言 key 不相同。这个细节你要是刚开始写装饰器大概率想不到但一旦线上出现诡异的数据错乱排查成本极高。9. 总结一些我踩过坑后沉淀下来的经验写这个 TTL 缓存装饰器最初只是为了解决一个线上接口响应慢的问题。后来在持续压测和代码演进中我逐渐意识到缓存装饰器远不是“查字典、存字典”这么简单。真正的难点在于如何保证缓存键的稳定性、如何处理并发环境下的竞态、如何控制缓存的资源消耗、如何让开发者直观地观测缓存的行为。最让我意外的是这个设计在项目里带来的不仅是性能提升还有一个隐藏的好处——它强制团队思考“哪些数据是可以容忍延迟的哪些数据必须实时读取”。以前大家写代码都是直接查数据库现在会先问一句“这个数据能缓存吗TTL 设多少合适”这种思维模式的改变比那几十毫秒的响应时间提升更有价值。如果你要在自己的项目里使用这套方案我的建议是从最简单的实现开始不要一上来就抄完整版。先用functools.lru_cache体验缓存的效果再替换成我第一版的memorize最后才上 TTL 和并发保护。每加一个特性都要明确它解决的是什么问题。这样即使以后出了问题你也能更快定位到具体的环节而不是在一堆复杂的代码里大海捞针。缓存是一把好用的双刃剑用得好性能翻倍用得不好数据一团乱关键在于你有没有真正理解它每个行为背后的代价。