先得说清楚我为什么会用“caveman”穴居人这么个梗来当项目代号。事情是这样的线上有个查询接口高峰期数据库连接数直接拉满告警我爬上去一看慢日志同一条SQL一秒钟被执行了上百次——同一个user_id同一个查询条件结果完全一样但每次请求都重新查了一遍库。当时团队第一反应是“加个Redis呗”但真去评估了一圈引入中间件、配置连接池、部署监控、缓存失效策略工程量一下子大得离谱为了这点事根本不值。于是我自己写了一个大概三十行的Python装饰器自动缓存函数结果、按时间过期、按大小淘汰零外部依赖往业务函数头上一挂就能用。这个装饰器的项目代号就叫caveman直译就是穴居人自嘲它原始、简单、但管用。如果你正在写Python服务想快速消除重复计算和重复SQL又不想因为这点需求就引入Redis这类外部中间件这篇文章应该能帮到你。就算你纯粹想自己动手实现一遍缓存的原理我下面会从设计思路到完整代码讲得比较透照着抄也问题不大。1. 别急着上Redis先搞清楚Caveman到底在解决什么问题1.1 问题的本质是“短时间相同输入得到相同输出”很多性能问题其实不用堆机器只要稍微想想数据的变化频率就够了。用户详情、商品信息、城市列表、字典表、地区编码……这些数据的真实变化频率是以分钟甚至小时为单位的可你的代码呢每收到一个请求就全量查一次数据库。这就像食堂阿姨今天菜单明明就那几个菜每来一个人她都要重新做一遍饭窗口围了一堆人师傅累得够呛出餐速度还慢。缓存要解决的本质上就是“短时间相同输入得到相同输出”的情况。把函数的入参作为缓存键把函数返回结果作为缓存值下次同样的参数进来就直接返回上次的结果不再重新执行函数体。这个逻辑说出来谁都能理解但真要落地的时候你会发现一堆细节等着你什么时候过期、缓存无限膨胀怎么办、并发环境下怎么保证不出错。当时我统计了一下手里的痛点场景基本都符合这个特征同一个用户信息的查询反复执行、某个统计报表五分钟内算出来的结果几乎不变、同一批商品详情的SQL被上下游服务反复调用。这些场景有一个共同点实时性要求没那么高允许几十秒甚至几分钟的数据延迟完全可以接受进程内缓存。1.2 缓存方案排序什么情况用什么招提到缓存很多人第一反应就是Redis这是对的但不是所有场景都有必要上。我在决定自己写之前把常见方案列了个表对比了一下。方案优点缺点适用场景Redis等外部缓存跨进程共享、容量大、可持久化运维成本高、有网络开销、需要额外部署监控多实例部署、需要分布式共享、数据量大且需要精细控制进程内缓存本项目零依赖、命中无网络开销、实现简单多进程不共享、容量受限于内存单机部署、读多写少、允许分钟级数据延迟不缓存无任何复杂度重复计算、数据库压力大、响应时间不稳定数据量极小或实时性要求极高我们当时的部署形态是单实例压根不存在多节点缓存同步的问题用进程内缓存是最合适的。执行函数拿到结果后直接塞进内存字典下次同样的参数进来直接查字典没有任何网络IO、没有序列化纯内存操作命中耗时基本都是微秒级。这个决定看起来简单但真正重要的是要知道这个方案的边界在哪如果后面做了多实例部署进程内缓存各缓存各的就不能继续这么干了得升级成Redis或者引入多级缓存方案那是后话。1.3 为什么叫“caveman”名字由来其实是自嘲。那天排查重复SQL查得烦了拍桌子说“这不就是原始人行为吗同一个问题反复解能不能让代码记住自己干过啥”。后来写出来的装饰器确实人如其名——它不会搞什么复杂的缓存预热、分布式锁、多级淘汰策略就是一个朴实的字典加过期时间像穴居人手里的石斧一样粗糙但有效。2. 核心设计拆解一个缓存装饰器该想清楚的几件事2.1 先写出能跑通的最小版本不管多复杂的方案我习惯先写一个最小可用的版本把核心机制跑通再一步步加保护。最简单的缓存装饰器长这样def caveman(func): cache {} def wrapper(*args, **kwargs): key (args, tuple(sorted(kwargs.items()))) if key not in cache: cache[key] func(*args, **kwargs) return cache[key] return wrapper用Python元组作为字典键把位置参数和关键字参数打包起来缓存结果。这个版本确实能工作但离“能上生产”还很远它永远不会过期一个key占用的内存永远不会释放它不管容量数据一多内存就爆它没有加锁多线程环境下同时写入同一个key可能会互相覆盖而且返回的结果可能会被调用方修改导致缓存被污染。不过别小看这个版本它把最核心的缓存机制讲清楚了。后面所有的升级本质都是在这个基础上做防御。2.2 三个必做的升级过期、容量、并发安全第一个是过期时间TTL。为什么需要过期因为业务数据总是在变的缓存如果不清理就会一直给你返回旧数据。实现方式很多我选择的是惰性过期——不启动后台定时扫描线程而是每次读取缓存的时候判断一下当前时间是否超过了过期时间。这么做的原因很简单省一个线程减少定时器带来的额外复杂度。时间戳的计算也不要用datetime对象直接用time.time()的浮点值比较快而且没有时区概念干扰。第二个是容量上限LRU。如果不限制容量缓存字典会无限增长。比如查询接口的参数是长尾的每天有几万个不同的user_id进来每个键存一份数据内存迟早被吃光。我用了Python自带的OrderedDict来实现LRULeast Recently Used淘汰每次命中或插入数据时用move_to_end把该键挪到最右边当缓存大小超过预设上限时用popitem(lastFalse)从最左边弹出最久没被访问的键。这个做法简洁有效不需要自己维护链表和哈希表的组合结构。第三个是并发安全。Python的字典操作虽然单个指令是原子的但“检查是否存在”“读取值”“写入值”这三个操作合在一起并不是原子的。多个线程同时命中同一个缓存键可能会在函数还没执行完的时候就有别的线程判断“缓存未命中”又去执行了一次真正的函数然后互相覆盖。最简单的方案是加锁我用的是threading.RLock()而不是threading.Lock()。RLock支持同一个线程重复获取锁这样即使后续某个缓存键的构建函数内部不小心又调了同一个被装饰的函数也不会把自己锁死。锁的覆盖范围也要小心。我强烈建议不要用锁把真正的函数调用包住因为函数执行往往涉及网络请求或数据库查询耗时不确定如果一直占着锁其他所有缓存读写都会被堵住。正确做法是锁只保护缓存字典的读写操作真正调用函数计算结果的阶段让锁放开。2.3 键的生成缓存命中的第一步缓存命中的前提是键必须稳定。两个容易踩的坑一是kwargs的顺序问题func(a1, b2)和func(b2, a1)在Python里语义完全相同但直接转成元组后键却不一样所以要把kwargs先排序再拼装二是不可哈希的问题某些参数类型比如列表、字典没法直接作为字典的键直接塞进去会抛异常。我的做法是生成一个字符串键把函数名、位置参数的repr、排序后的关键字参数拼起来。这样做对性能的影响几乎可以忽略而且调试起来也方便。有一点要提醒repr的结果在不同Python版本间理论上可能有差异但对绝大多数业务场景来说这个差异不会造成实际问题。如果你要做得更严格可以考虑用pickle库的协议来序列化参数但那又引入了其他限制。另外有个问题容易被忽视两个不同函数如果参数完全一样它们生成的键会不会冲突解决方式很简单把函数名放进键里面就行。3. 完整代码与实操从零跑起一个Caveman3.1 完整实现代码下面是完整的装饰器代码我用它搭了两个线上服务稳定跑到现在没什么问题。import functools import threading import time from collections import OrderedDict def caveman(expire60, maxsize1024): def decorator(func): cache OrderedDict() lock threading.RLock() def make_key(args, kwargs): parts [func.__name__] parts.extend(repr(arg) for arg in args) parts.extend(f{k}{v!r} for k, v in sorted(kwargs.items())) return |.join(parts) functools.wraps(func) def wrapper(*args, **kwargs): key make_key(args, kwargs) now time.time() with lock: if key in cache: deadline, result cache[key] if deadline now: cache.move_to_end(key) return result del cache[key] result func(*args, **kwargs) with lock: cache[key] (now expire, result) cache.move_to_end(key) while len(cache) maxsize: cache.popitem(lastFalse) return result def cache_clear(): with lock: cache.clear() def cache_info(): return {size: len(cache), maxsize: maxsize, expire: expire} wrapper.cache_clear cache_clear wrapper.cache_info cache_info return wrapper return decorator逐段说一下关键逻辑。make_key负责生成稳定键把函数名放进键里解决同名函数的冲突问题。读取缓存时先在锁内判断键是否存在如果存在再判断过期时间没过期就把键移到最右边并返回缓存值如果过期了就删掉这个键。真正执行函数结果时不对缓存字典持锁这是避免并发性能瓶颈的关键。执行完拿到结果后回写缓存并检查容量超出上限就从最左边弹出最久未使用的键。cache_clear和cache_info这两个方法是隐藏的彩蛋挂在wrapper函数上方便线上调试。functools.wraps不是摆设它能保留被装饰函数的__name__、__doc__等元信息调试和文档生成的时候会省很多心。3.2 基本用法装饰无参函数、有参函数、对象方法用法很简单直接往函数头上挂就行caveman(expire30) def get_city_name(code): # 假设这里查数据库 time.sleep(0.1) return fcity-{code} caveman(expire5, maxsize64) def get_discount(user_level, product_id1): return user_level * product_id无参函数同样可以装饰只是所有调用都命中同一个缓存。对象方法需要注意一个点如果方法内部结果依赖于self的状态那么self的变化也会影响结果这时候要么把状态字段作为参数单独传进去参与缓存键要么干脆不要在这种场景用缓存。比如class UserService: def __init__(self, config): self.config config caveman(expire10) def get_point(self, user_id): return self.config.get(point_base) user_id这种写法就有隐患因为config如果中途变化了缓存的结果不会跟着变。建议的方法是不要装饰在依赖实例状态的方法上而是提示一层独立函数把依赖的状态作为参数传进去。3.3 expire和maxsize的实战配置建议这两个参数关系到缓存是否真的有效。expire应该根据数据的变化频率来定如果数据每隔一分钟会变那expire30~60秒是比较合适的如果对实时性要求高比如库存数量就别用缓存哪怕缓存一秒都可能出问题如果是统计报表这种几乎不变的expire300都不过分。基本思路是取“业务能接受的最长延迟”作为过期时间不要拍脑袋写个极小值那样缓存命中率会很低等于白上。maxsize则取决于你能接受的内存上限和真实的唯一参数组合数。简单估算每条缓存结果平均多大乘以预估的唯一key数量再乘以2的余量系数看是否在可控范围内。比如你的函数结果平均10KB预估最多1000个不同参数组合那大概就是10MB内存加一倍余量也就20MB这个开销绝大多数服务都能接受。但如果你的key组合是海量的比如每天10万个唯一查询参数那就要重新考虑方案了。还有一个容易被忽略的原则写操作不要缓存查询操作里面如果混入了随机值也不要缓存。比如time.time()、UUID作为参数传进函数缓存基本就废了因为每个请求都是新key。4. 真实接入案例一个用户信息接口的完整改造过程4.1 改造前后的代码对比当时我们有个Flask写的用户信息接口每次请求都会执行一次完整的数据库查询高峰期数据库连接被打满。改造前核心逻辑大概长这样app.get(/api/user) def user_detail(): user_id request.args.get(user_id) row db.query(SELECT * FROM user WHERE id?, user_id) return {user: row}问题很明显query直接用在了视图函数里相同的user_id来了多少次数据库就被查多少次。我改造时做了一件事把“查用户数据”这个纯函数单独拎出来给它加上caveman(expire30, maxsize512)然后让视图函数调用它。caveman(expire30, maxsize512) def load_user(user_id): row db.query(SELECT * FROM user WHERE id?, user_id) return row app.get(/api/user) def user_detail(): user_id request.args.get(user_id) return {user: load_user(user_id)}为什么不直接在视图函数上加缓存因为视图函数除了真正的查询参数还依赖request这种上下文对象直接装饰它会导致缓存键里混入无意义的请求对象结果永远无法命中。把核心查询逻辑提取成独立函数天然对标“相同输入得到相同输出”才是缓存能发挥作用的正确姿势。4.2 验证效果用数字说服人加完缓存不能只靠感觉说“好像快了”得拿数字出来。我写了个简单的对比脚本import time caveman(expire60) def expensive(x): time.sleep(0.05) return x * 2 st time.time() for _ in range(1000): expensive(42) print(with cache cost:, time.time() - st)这个函数每次执行要睡50毫秒模拟数据库查询耗时。1000次调用如果不加缓存理论耗时是50秒加了缓存后只有第一次调用真正执行了函数后面999次都在微秒级命中整个循环几乎瞬间跑完。实际数字拍出来群里的人自然就服气了。上线之后我又看了两天监控接口P99响应时间从180毫秒直接降到20毫秒以内数据库连接数也稳定了。4.3 一个容易踩的坑缓存了可变对象这里有个非常隐蔽的坑很多人第一次用都栽了。如果缓存的函数返回的是list、dict这种可变对象调用方拿到的其实是缓存里的同一个引用一旦调用方改了返回值缓存也就被污染了。比如caveman(expire60) def get_tags(): return [a, b] tags get_tags() tags.append(c) # 下次再调用 get_tags()返回的是 [a,b,c]这个问题处理方式有三种一是尽量让缓存函数返回不可变类型比如把list改成tuple把dict改成换一种不可变表示二是在文档或注释里明确写明“不要修改返回值”三是在返回之前用copy.deepcopy拷贝一份让调用方改的是副本代价是深拷贝本身有开销不适合大对象。我自己的习惯是优先用不可变类型和文档约束只有对安全要求极高的场景才考虑深拷贝。5. 常见问题与排查技巧实录5.1 缓存命中率低先看你的key里是不是有随机参数我见过很多“上了缓存但接口还是很慢”的案例排到最后发现是key的问题。比如视图函数里把request.full_path当参数传给了缓存函数路径里带了?from1234567890这种时间戳参数每个请求都是一个新key缓存自然永远不命中。排查技巧很简单在make_key里临时打印一下生成的key人为调用两次相同参数的函数看看key是否一致。如果打印出来的key每次都不同就沿着参数排查把随机参数和无关上下文从缓存函数的入参列表里剔除。5.2 缓存穿透大量请求同时未命中全部打到数据库缓存穿透是指一个缓存key刚过期此时突然涌入大量相同参数的请求它们同时发现缓存未命中于是每个线程都执行了一遍真正的查询函数数据库压力瞬间飙升。解决思路是“单飞”single flight——同一时刻同一个key只允许一个线程去真正执行函数其他线程阻塞等待结果。我在这套装饰器里没有默认内置单飞目的是让核心保持简单但需要的时候可以组合使用。伪代码大致是这样inflight {} build_lock threading.Lock() def build(key): with build_lock: if key not in inflight: inflight[key] threading.Event() # 第一个线程执行真正的函数并set event # 其他线程等待event实际实现会稍微复杂要考虑异常情况下event永远不触发的处理。对于大多数业务场景如果数据库本身有连接池保护缓存穿透的后果没有想象中严重真正需要单飞的是下游响应特别慢、而且并发量特别大的场景。5.3 缓存雪崩同一秒所有key一起过期如果所有缓存的expire都是同一个固定值那么它们的过期时间会集中在同一时刻。此时如果请求量很大所有未命中的请求会同时冲向后端。解决办法是给过期时间加一个随机抖动import random jitter random.uniform(0.9, 1.1) cache[key] (now expire * jitter, result)这样每个key的实际过期时间分布在expire附近不会整整齐齐地一起过期。抖动范围我建议控制在±10%左右太大会影响缓存的平均过期时间太小又起不到打散的效果。这招不需要写进每一份代码里但在大规模接入时一定要考虑。5.4 内存悄悄上涨maxsize和过期策略失效的后果进程内缓存最怕的就是内存泄漏。虽然有了maxsize限制但有一个细节容易忽略数据过期后在它被下次读取之前那个键仍然占着内存。如果key特别多过期键堆积起来的内存就不能小看。比如maxsize设为1024但实际生产环境每天产生10万个唯一keyLRU淘汰会把这些没用的键弹出去看似没问题但如果淘汰速度跟不上写入速度字典长度会一直顶在maxsize上所有键都活得不久缓存命中率又很低等于缓存没啥用还占着内存。我的排查方法是定期调wrapper.cache_info()看一下size字段如果长期顶在maxsize附近说明不是缓存命中率高而是淘汰在疯狂发生需要调大maxsize或reduce key的独特参数数量。另一个办法是给装饰器加一个定时清理过期键的功能但那会增加复杂度我建议不要轻易去实现先用cache_info定位问题再决定方案。5.5 数据主动变更了缓存里的旧数据怎么办这是很多人问得最多的点后台明明改了数据接口返回的却是旧值。答案要分场景。如果允许分钟级延迟最省心的做法就是把expire调小让缓存自然过期比如设为5~10秒用户几乎感知不到延迟代码也不用做任何额外处理。如果数据一致性要求比较高可以在写操作完成之后调用wrapper.cache_clear()整体清空缓存——这个操作简单粗暴代价是所有缓存键都会被清掉下一波请求会有一小段短暂的回源压力但通常问题不大。真正复杂的做法是针对性删除特定key也就是主动失效某一个缓存键。这件事看起来容易做起来要非常小心因为你需要能精确复现key的生成逻辑哪怕一个参数的repr格式对不上删除就会失败。我自己的经验是在绝大多数业务系统里用“调小expire”和“整体清空”两个手段就够了主动删指定key适合的场景极少还容易引入一堆bug。6. 最后再聊聊我的几点实际体会工具越是小越是简单越需要按生产标准来写。我最初写caveman的时候只用了30行后来加了RLock、LRU、过期清理、cache_info代码量翻了一倍多但每一行都是线上踩坑踩出来的。加锁不是为了好看是因为多线程并发真的会互相覆盖限制容量不是为了装是因为真的有人会遇到内存疯涨。不要以为“简单的东西不需要防御”简单的是使用姿势不是防御标准。关于要不要上Redis我的看法一直很务实先测量再决定。如果一个查询结果的生成成本是几百毫秒级别重复率又高那么进程内缓存带来的收益已经很可观完全没有必要引入中间件。反过来如果你已经知道服务是多实例部署缓存必须跨进程共享那从一开始就该评估Redis之类的方案别先用进程内缓存临时糊着后面再改架构更痛苦。测一测你的热路径算一算重复查询的量级答案自然就出来了。还有一个小技巧值得分享给缓存工具挂上cache_info和cache_clear这两个方法。线上出了问题你先调一下cache_info()看看size是不是异常、maxsize是不是合理可能比看半天日志都快。缓存的难点从来不在“写入”和“命中”而在“什么时候失效、占多少内存、并发怎么保护”——把这些想清楚你写的任何一个小工具都能稳稳跑在生产环境里。 
长河君 · 2026/10/29 6:12:49 · 阅读 1.2k 用30行Python装饰器实现进程内缓存，替代Redis的轻量级方案
先得说清楚我为什么会用“caveman”穴居人这么个梗来当项目代号。事情是这样的线上有个查询接口高峰期数据库连接数直接拉满告警我爬上去一看慢日志同一条SQL一秒钟被执行了上百次——同一个user_id同一个查询条件结果完全一样但每次请求都重新查了一遍库。当时团队第一反应是“加个Redis呗”但真去评估了一圈引入中间件、配置连接池、部署监控、缓存失效策略工程量一下子大得离谱为了这点事根本不值。于是我自己写了一个大概三十行的Python装饰器自动缓存函数结果、按时间过期、按大小淘汰零外部依赖往业务函数头上一挂就能用。这个装饰器的项目代号就叫caveman直译就是穴居人自嘲它原始、简单、但管用。如果你正在写Python服务想快速消除重复计算和重复SQL又不想因为这点需求就引入Redis这类外部中间件这篇文章应该能帮到你。就算你纯粹想自己动手实现一遍缓存的原理我下面会从设计思路到完整代码讲得比较透照着抄也问题不大。1. 别急着上Redis先搞清楚Caveman到底在解决什么问题1.1 问题的本质是“短时间相同输入得到相同输出”很多性能问题其实不用堆机器只要稍微想想数据的变化频率就够了。用户详情、商品信息、城市列表、字典表、地区编码……这些数据的真实变化频率是以分钟甚至小时为单位的可你的代码呢每收到一个请求就全量查一次数据库。这就像食堂阿姨今天菜单明明就那几个菜每来一个人她都要重新做一遍饭窗口围了一堆人师傅累得够呛出餐速度还慢。缓存要解决的本质上就是“短时间相同输入得到相同输出”的情况。把函数的入参作为缓存键把函数返回结果作为缓存值下次同样的参数进来就直接返回上次的结果不再重新执行函数体。这个逻辑说出来谁都能理解但真要落地的时候你会发现一堆细节等着你什么时候过期、缓存无限膨胀怎么办、并发环境下怎么保证不出错。当时我统计了一下手里的痛点场景基本都符合这个特征同一个用户信息的查询反复执行、某个统计报表五分钟内算出来的结果几乎不变、同一批商品详情的SQL被上下游服务反复调用。这些场景有一个共同点实时性要求没那么高允许几十秒甚至几分钟的数据延迟完全可以接受进程内缓存。1.2 缓存方案排序什么情况用什么招提到缓存很多人第一反应就是Redis这是对的但不是所有场景都有必要上。我在决定自己写之前把常见方案列了个表对比了一下。方案优点缺点适用场景Redis等外部缓存跨进程共享、容量大、可持久化运维成本高、有网络开销、需要额外部署监控多实例部署、需要分布式共享、数据量大且需要精细控制进程内缓存本项目零依赖、命中无网络开销、实现简单多进程不共享、容量受限于内存单机部署、读多写少、允许分钟级数据延迟不缓存无任何复杂度重复计算、数据库压力大、响应时间不稳定数据量极小或实时性要求极高我们当时的部署形态是单实例压根不存在多节点缓存同步的问题用进程内缓存是最合适的。执行函数拿到结果后直接塞进内存字典下次同样的参数进来直接查字典没有任何网络IO、没有序列化纯内存操作命中耗时基本都是微秒级。这个决定看起来简单但真正重要的是要知道这个方案的边界在哪如果后面做了多实例部署进程内缓存各缓存各的就不能继续这么干了得升级成Redis或者引入多级缓存方案那是后话。1.3 为什么叫“caveman”名字由来其实是自嘲。那天排查重复SQL查得烦了拍桌子说“这不就是原始人行为吗同一个问题反复解能不能让代码记住自己干过啥”。后来写出来的装饰器确实人如其名——它不会搞什么复杂的缓存预热、分布式锁、多级淘汰策略就是一个朴实的字典加过期时间像穴居人手里的石斧一样粗糙但有效。2. 核心设计拆解一个缓存装饰器该想清楚的几件事2.1 先写出能跑通的最小版本不管多复杂的方案我习惯先写一个最小可用的版本把核心机制跑通再一步步加保护。最简单的缓存装饰器长这样def caveman(func): cache {} def wrapper(*args, **kwargs): key (args, tuple(sorted(kwargs.items()))) if key not in cache: cache[key] func(*args, **kwargs) return cache[key] return wrapper用Python元组作为字典键把位置参数和关键字参数打包起来缓存结果。这个版本确实能工作但离“能上生产”还很远它永远不会过期一个key占用的内存永远不会释放它不管容量数据一多内存就爆它没有加锁多线程环境下同时写入同一个key可能会互相覆盖而且返回的结果可能会被调用方修改导致缓存被污染。不过别小看这个版本它把最核心的缓存机制讲清楚了。后面所有的升级本质都是在这个基础上做防御。2.2 三个必做的升级过期、容量、并发安全第一个是过期时间TTL。为什么需要过期因为业务数据总是在变的缓存如果不清理就会一直给你返回旧数据。实现方式很多我选择的是惰性过期——不启动后台定时扫描线程而是每次读取缓存的时候判断一下当前时间是否超过了过期时间。这么做的原因很简单省一个线程减少定时器带来的额外复杂度。时间戳的计算也不要用datetime对象直接用time.time()的浮点值比较快而且没有时区概念干扰。第二个是容量上限LRU。如果不限制容量缓存字典会无限增长。比如查询接口的参数是长尾的每天有几万个不同的user_id进来每个键存一份数据内存迟早被吃光。我用了Python自带的OrderedDict来实现LRULeast Recently Used淘汰每次命中或插入数据时用move_to_end把该键挪到最右边当缓存大小超过预设上限时用popitem(lastFalse)从最左边弹出最久没被访问的键。这个做法简洁有效不需要自己维护链表和哈希表的组合结构。第三个是并发安全。Python的字典操作虽然单个指令是原子的但“检查是否存在”“读取值”“写入值”这三个操作合在一起并不是原子的。多个线程同时命中同一个缓存键可能会在函数还没执行完的时候就有别的线程判断“缓存未命中”又去执行了一次真正的函数然后互相覆盖。最简单的方案是加锁我用的是threading.RLock()而不是threading.Lock()。RLock支持同一个线程重复获取锁这样即使后续某个缓存键的构建函数内部不小心又调了同一个被装饰的函数也不会把自己锁死。锁的覆盖范围也要小心。我强烈建议不要用锁把真正的函数调用包住因为函数执行往往涉及网络请求或数据库查询耗时不确定如果一直占着锁其他所有缓存读写都会被堵住。正确做法是锁只保护缓存字典的读写操作真正调用函数计算结果的阶段让锁放开。2.3 键的生成缓存命中的第一步缓存命中的前提是键必须稳定。两个容易踩的坑一是kwargs的顺序问题func(a1, b2)和func(b2, a1)在Python里语义完全相同但直接转成元组后键却不一样所以要把kwargs先排序再拼装二是不可哈希的问题某些参数类型比如列表、字典没法直接作为字典的键直接塞进去会抛异常。我的做法是生成一个字符串键把函数名、位置参数的repr、排序后的关键字参数拼起来。这样做对性能的影响几乎可以忽略而且调试起来也方便。有一点要提醒repr的结果在不同Python版本间理论上可能有差异但对绝大多数业务场景来说这个差异不会造成实际问题。如果你要做得更严格可以考虑用pickle库的协议来序列化参数但那又引入了其他限制。另外有个问题容易被忽视两个不同函数如果参数完全一样它们生成的键会不会冲突解决方式很简单把函数名放进键里面就行。3. 完整代码与实操从零跑起一个Caveman3.1 完整实现代码下面是完整的装饰器代码我用它搭了两个线上服务稳定跑到现在没什么问题。import functools import threading import time from collections import OrderedDict def caveman(expire60, maxsize1024): def decorator(func): cache OrderedDict() lock threading.RLock() def make_key(args, kwargs): parts [func.__name__] parts.extend(repr(arg) for arg in args) parts.extend(f{k}{v!r} for k, v in sorted(kwargs.items())) return |.join(parts) functools.wraps(func) def wrapper(*args, **kwargs): key make_key(args, kwargs) now time.time() with lock: if key in cache: deadline, result cache[key] if deadline now: cache.move_to_end(key) return result del cache[key] result func(*args, **kwargs) with lock: cache[key] (now expire, result) cache.move_to_end(key) while len(cache) maxsize: cache.popitem(lastFalse) return result def cache_clear(): with lock: cache.clear() def cache_info(): return {size: len(cache), maxsize: maxsize, expire: expire} wrapper.cache_clear cache_clear wrapper.cache_info cache_info return wrapper return decorator逐段说一下关键逻辑。make_key负责生成稳定键把函数名放进键里解决同名函数的冲突问题。读取缓存时先在锁内判断键是否存在如果存在再判断过期时间没过期就把键移到最右边并返回缓存值如果过期了就删掉这个键。真正执行函数结果时不对缓存字典持锁这是避免并发性能瓶颈的关键。执行完拿到结果后回写缓存并检查容量超出上限就从最左边弹出最久未使用的键。cache_clear和cache_info这两个方法是隐藏的彩蛋挂在wrapper函数上方便线上调试。functools.wraps不是摆设它能保留被装饰函数的__name__、__doc__等元信息调试和文档生成的时候会省很多心。3.2 基本用法装饰无参函数、有参函数、对象方法用法很简单直接往函数头上挂就行caveman(expire30) def get_city_name(code): # 假设这里查数据库 time.sleep(0.1) return fcity-{code} caveman(expire5, maxsize64) def get_discount(user_level, product_id1): return user_level * product_id无参函数同样可以装饰只是所有调用都命中同一个缓存。对象方法需要注意一个点如果方法内部结果依赖于self的状态那么self的变化也会影响结果这时候要么把状态字段作为参数单独传进去参与缓存键要么干脆不要在这种场景用缓存。比如class UserService: def __init__(self, config): self.config config caveman(expire10) def get_point(self, user_id): return self.config.get(point_base) user_id这种写法就有隐患因为config如果中途变化了缓存的结果不会跟着变。建议的方法是不要装饰在依赖实例状态的方法上而是提示一层独立函数把依赖的状态作为参数传进去。3.3 expire和maxsize的实战配置建议这两个参数关系到缓存是否真的有效。expire应该根据数据的变化频率来定如果数据每隔一分钟会变那expire30~60秒是比较合适的如果对实时性要求高比如库存数量就别用缓存哪怕缓存一秒都可能出问题如果是统计报表这种几乎不变的expire300都不过分。基本思路是取“业务能接受的最长延迟”作为过期时间不要拍脑袋写个极小值那样缓存命中率会很低等于白上。maxsize则取决于你能接受的内存上限和真实的唯一参数组合数。简单估算每条缓存结果平均多大乘以预估的唯一key数量再乘以2的余量系数看是否在可控范围内。比如你的函数结果平均10KB预估最多1000个不同参数组合那大概就是10MB内存加一倍余量也就20MB这个开销绝大多数服务都能接受。但如果你的key组合是海量的比如每天10万个唯一查询参数那就要重新考虑方案了。还有一个容易被忽略的原则写操作不要缓存查询操作里面如果混入了随机值也不要缓存。比如time.time()、UUID作为参数传进函数缓存基本就废了因为每个请求都是新key。4. 真实接入案例一个用户信息接口的完整改造过程4.1 改造前后的代码对比当时我们有个Flask写的用户信息接口每次请求都会执行一次完整的数据库查询高峰期数据库连接被打满。改造前核心逻辑大概长这样app.get(/api/user) def user_detail(): user_id request.args.get(user_id) row db.query(SELECT * FROM user WHERE id?, user_id) return {user: row}问题很明显query直接用在了视图函数里相同的user_id来了多少次数据库就被查多少次。我改造时做了一件事把“查用户数据”这个纯函数单独拎出来给它加上caveman(expire30, maxsize512)然后让视图函数调用它。caveman(expire30, maxsize512) def load_user(user_id): row db.query(SELECT * FROM user WHERE id?, user_id) return row app.get(/api/user) def user_detail(): user_id request.args.get(user_id) return {user: load_user(user_id)}为什么不直接在视图函数上加缓存因为视图函数除了真正的查询参数还依赖request这种上下文对象直接装饰它会导致缓存键里混入无意义的请求对象结果永远无法命中。把核心查询逻辑提取成独立函数天然对标“相同输入得到相同输出”才是缓存能发挥作用的正确姿势。4.2 验证效果用数字说服人加完缓存不能只靠感觉说“好像快了”得拿数字出来。我写了个简单的对比脚本import time caveman(expire60) def expensive(x): time.sleep(0.05) return x * 2 st time.time() for _ in range(1000): expensive(42) print(with cache cost:, time.time() - st)这个函数每次执行要睡50毫秒模拟数据库查询耗时。1000次调用如果不加缓存理论耗时是50秒加了缓存后只有第一次调用真正执行了函数后面999次都在微秒级命中整个循环几乎瞬间跑完。实际数字拍出来群里的人自然就服气了。上线之后我又看了两天监控接口P99响应时间从180毫秒直接降到20毫秒以内数据库连接数也稳定了。4.3 一个容易踩的坑缓存了可变对象这里有个非常隐蔽的坑很多人第一次用都栽了。如果缓存的函数返回的是list、dict这种可变对象调用方拿到的其实是缓存里的同一个引用一旦调用方改了返回值缓存也就被污染了。比如caveman(expire60) def get_tags(): return [a, b] tags get_tags() tags.append(c) # 下次再调用 get_tags()返回的是 [a,b,c]这个问题处理方式有三种一是尽量让缓存函数返回不可变类型比如把list改成tuple把dict改成换一种不可变表示二是在文档或注释里明确写明“不要修改返回值”三是在返回之前用copy.deepcopy拷贝一份让调用方改的是副本代价是深拷贝本身有开销不适合大对象。我自己的习惯是优先用不可变类型和文档约束只有对安全要求极高的场景才考虑深拷贝。5. 常见问题与排查技巧实录5.1 缓存命中率低先看你的key里是不是有随机参数我见过很多“上了缓存但接口还是很慢”的案例排到最后发现是key的问题。比如视图函数里把request.full_path当参数传给了缓存函数路径里带了?from1234567890这种时间戳参数每个请求都是一个新key缓存自然永远不命中。排查技巧很简单在make_key里临时打印一下生成的key人为调用两次相同参数的函数看看key是否一致。如果打印出来的key每次都不同就沿着参数排查把随机参数和无关上下文从缓存函数的入参列表里剔除。5.2 缓存穿透大量请求同时未命中全部打到数据库缓存穿透是指一个缓存key刚过期此时突然涌入大量相同参数的请求它们同时发现缓存未命中于是每个线程都执行了一遍真正的查询函数数据库压力瞬间飙升。解决思路是“单飞”single flight——同一时刻同一个key只允许一个线程去真正执行函数其他线程阻塞等待结果。我在这套装饰器里没有默认内置单飞目的是让核心保持简单但需要的时候可以组合使用。伪代码大致是这样inflight {} build_lock threading.Lock() def build(key): with build_lock: if key not in inflight: inflight[key] threading.Event() # 第一个线程执行真正的函数并set event # 其他线程等待event实际实现会稍微复杂要考虑异常情况下event永远不触发的处理。对于大多数业务场景如果数据库本身有连接池保护缓存穿透的后果没有想象中严重真正需要单飞的是下游响应特别慢、而且并发量特别大的场景。5.3 缓存雪崩同一秒所有key一起过期如果所有缓存的expire都是同一个固定值那么它们的过期时间会集中在同一时刻。此时如果请求量很大所有未命中的请求会同时冲向后端。解决办法是给过期时间加一个随机抖动import random jitter random.uniform(0.9, 1.1) cache[key] (now expire * jitter, result)这样每个key的实际过期时间分布在expire附近不会整整齐齐地一起过期。抖动范围我建议控制在±10%左右太大会影响缓存的平均过期时间太小又起不到打散的效果。这招不需要写进每一份代码里但在大规模接入时一定要考虑。5.4 内存悄悄上涨maxsize和过期策略失效的后果进程内缓存最怕的就是内存泄漏。虽然有了maxsize限制但有一个细节容易忽略数据过期后在它被下次读取之前那个键仍然占着内存。如果key特别多过期键堆积起来的内存就不能小看。比如maxsize设为1024但实际生产环境每天产生10万个唯一keyLRU淘汰会把这些没用的键弹出去看似没问题但如果淘汰速度跟不上写入速度字典长度会一直顶在maxsize上所有键都活得不久缓存命中率又很低等于缓存没啥用还占着内存。我的排查方法是定期调wrapper.cache_info()看一下size字段如果长期顶在maxsize附近说明不是缓存命中率高而是淘汰在疯狂发生需要调大maxsize或reduce key的独特参数数量。另一个办法是给装饰器加一个定时清理过期键的功能但那会增加复杂度我建议不要轻易去实现先用cache_info定位问题再决定方案。5.5 数据主动变更了缓存里的旧数据怎么办这是很多人问得最多的点后台明明改了数据接口返回的却是旧值。答案要分场景。如果允许分钟级延迟最省心的做法就是把expire调小让缓存自然过期比如设为5~10秒用户几乎感知不到延迟代码也不用做任何额外处理。如果数据一致性要求比较高可以在写操作完成之后调用wrapper.cache_clear()整体清空缓存——这个操作简单粗暴代价是所有缓存键都会被清掉下一波请求会有一小段短暂的回源压力但通常问题不大。真正复杂的做法是针对性删除特定key也就是主动失效某一个缓存键。这件事看起来容易做起来要非常小心因为你需要能精确复现key的生成逻辑哪怕一个参数的repr格式对不上删除就会失败。我自己的经验是在绝大多数业务系统里用“调小expire”和“整体清空”两个手段就够了主动删指定key适合的场景极少还容易引入一堆bug。6. 最后再聊聊我的几点实际体会工具越是小越是简单越需要按生产标准来写。我最初写caveman的时候只用了30行后来加了RLock、LRU、过期清理、cache_info代码量翻了一倍多但每一行都是线上踩坑踩出来的。加锁不是为了好看是因为多线程并发真的会互相覆盖限制容量不是为了装是因为真的有人会遇到内存疯涨。不要以为“简单的东西不需要防御”简单的是使用姿势不是防御标准。关于要不要上Redis我的看法一直很务实先测量再决定。如果一个查询结果的生成成本是几百毫秒级别重复率又高那么进程内缓存带来的收益已经很可观完全没有必要引入中间件。反过来如果你已经知道服务是多实例部署缓存必须跨进程共享那从一开始就该评估Redis之类的方案别先用进程内缓存临时糊着后面再改架构更痛苦。测一测你的热路径算一算重复查询的量级答案自然就出来了。还有一个小技巧值得分享给缓存工具挂上cache_info和cache_clear这两个方法。线上出了问题你先调一下cache_info()看看size是不是异常、maxsize是不是合理可能比看半天日志都快。缓存的难点从来不在“写入”和“命中”而在“什么时候失效、占多少内存、并发怎么保护”——把这些想清楚你写的任何一个小工具都能稳稳跑在生产环境里。