企业官网建设流程全解析

先得说清楚，我为什么会用“caveman”（穴居人）这么个梗来当项目代号。事情是这样的：线上有个查询接口，高峰期数据库连接数直接拉满告警，我爬上去一看慢日志，同一条SQL一秒钟被执行了上百次——同一个user_id，同一个查询条件，结果完全一样，但每次请求都重新查了一遍库。当时团队第一反应是“加个Redis呗”，但真去评估了一圈，引入中间件、配置连接池、部署监控、缓存失效策略，工程量一下子大得离谱，为了这点事根本不值。于是我自己写了一个大概三十行的Python装饰器，自动缓存函数结果、按时间过期、按大小淘汰，零外部依赖，往业务函数头上一挂就能用。这个装饰器的项目代号就叫caveman，直译就是穴居人，自嘲它原始、简单、但管用。

如果你正在写Python服务，想快速消除重复计算和重复SQL，又不想因为这点需求就引入Redis这类外部中间件，这篇文章应该能帮到你。就算你纯粹想自己动手实现一遍缓存的原理，我下面会从设计思路到完整代码讲得比较透，照着抄也问题不大。

1. 别急着上Redis：先搞清楚Caveman到底在解决什么问题

1.1 问题的本质是“短时间相同输入得到相同输出”

很多性能问题其实不用堆机器，只要稍微想想数据的变化频率就够了。用户详情、商品信息、城市列表、字典表、地区编码……这些数据的真实变化频率是以分钟甚至小时为单位的，可你的代码呢？每收到一个请求就全量查一次数据库。这就像食堂阿姨，今天菜单明明就那几个菜，每来一个人她都要重新做一遍饭，窗口围了一堆人，师傅累得够呛，出餐速度还慢。

缓存要解决的，本质上就是“短时间相同输入得到相同输出”的情况。把函数的入参作为缓存键，把函数返回结果作为缓存值，下次同样的参数进来就直接返回上次的结果，不再重新执行函数体。这个逻辑说出来谁都能理解，但真要落地的时候，你会发现一堆细节等着你：什么时候过期、缓存无限膨胀怎么办、并发环境下怎么保证不出错。

当时我统计了一下手里的痛点场景，基本都符合这个特征：同一个用户信息的查询反复执行、某个统计报表五分钟内算出来的结果几乎不变、同一批商品详情的SQL被上下游服务反复调用。这些场景有一个共同点，实时性要求没那么高，允许几十秒甚至几分钟的数据延迟，完全可以接受进程内缓存。

1.2 缓存方案排序：什么情况用什么招

提到缓存，很多人第一反应就是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来实现LRU（Least Recently Used）淘汰：每次命中或插入数据时，用move_to_end把该键挪到最右边；当缓存大小超过预设上限时，用popitem(last=False)从最左边弹出最久没被访问的键。这个做法简洁有效，不需要自己维护链表和哈希表的组合结构。

第三个是并发安全。Python的字典操作虽然单个指令是原子的，但“检查是否存在”“读取值”“写入值”这三个操作合在一起并不是原子的。多个线程同时命中同一个缓存键，可能会在函数还没执行完的时候就有别的线程判断“缓存未命中”，又去执行了一次真正的函数，然后互相覆盖。最简单的方案是加锁，我用的是threading.RLock()而不是threading.Lock()。RLock支持同一个线程重复获取锁，这样即使后续某个缓存键的构建函数内部不小心又调了同一个被装饰的函数，也不会把自己锁死。

锁的覆盖范围也要小心。我强烈建议不要用锁把真正的函数调用包住，因为函数执行往往涉及网络请求或数据库查询，耗时不确定，如果一直占着锁，其他所有缓存读写都会被堵住。正确做法是：锁只保护缓存字典的读写操作，真正调用函数计算结果的阶段让锁放开。

2.3 键的生成：缓存命中的第一步

缓存命中的前提是键必须稳定。两个容易踩的坑：一是kwargs的顺序问题，func(a=1, b=2)和func(b=2, a=1)在Python里语义完全相同，但直接转成元组后键却不一样，所以要把kwargs先排序再拼装；二是不可哈希的问题，某些参数类型比如列表、字典没法直接作为字典的键，直接塞进去会抛异常。

我的做法是生成一个字符串键，把函数名、位置参数的repr、排序后的关键字参数拼起来。这样做对性能的影响几乎可以忽略，而且调试起来也方便。有一点要提醒：repr的结果在不同Python版本间理论上可能有差异，但对绝大多数业务场景来说这个差异不会造成实际问题。如果你要做得更严格，可以考虑用pickle库的协议来序列化参数，但那又引入了其他限制。

另外有个问题容易被忽视：两个不同函数，如果参数完全一样，它们生成的键会不会冲突？解决方式很简单，把函数名放进键里面就行。

3. 完整代码与实操：从零跑起一个Caveman

3.1 完整实现代码

下面是完整的装饰器代码，我用它搭了两个线上服务，稳定跑到现在没什么问题。

import functools import threading import time from collections import OrderedDict def caveman(expire=60, maxsize=1024): 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(last=False) 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(expire=30) def get_city_name(code): # 假设这里查数据库 time.sleep(0.1) return f"city-{code}" @caveman(expire=5, maxsize=64) def get_discount(user_level, product_id=1): return user_level * product_id

无参函数同样可以装饰，只是所有调用都命中同一个缓存。对象方法需要注意一个点：如果方法内部结果依赖于self的状态，那么self的变化也会影响结果，这时候要么把状态字段作为参数单独传进去参与缓存键，要么干脆不要在这种场景用缓存。比如：

class UserService: def __init__(self, config): self.config = config @caveman(expire=10) def get_point(self, user_id): return self.config.get("point_base") + user_id

这种写法就有隐患，因为config如果中途变化了，缓存的结果不会跟着变。建议的方法是不要装饰在依赖实例状态的方法上，而是提示一层独立函数，把依赖的状态作为参数传进去。

3.3 expire和maxsize的实战配置建议

这两个参数关系到缓存是否真的有效。expire应该根据数据的变化频率来定：如果数据每隔一分钟会变，那expire=30~60秒是比较合适的；如果对实时性要求高，比如库存数量，就别用缓存，哪怕缓存一秒都可能出问题；如果是统计报表这种几乎不变的，expire=300都不过分。基本思路是：取“业务能接受的最长延迟”作为过期时间，不要拍脑袋写个极小值，那样缓存命中率会很低，等于白上。

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(expire=30, maxsize=512)，然后让视图函数调用它。

@caveman(expire=30, maxsize=512) 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(expire=60) 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(expire=60) 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当参数传给了缓存函数，路径里带了?from=1234567890这种时间戳参数，每个请求都是一个新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万个唯一key，LRU淘汰会把这些没用的键弹出去，看似没问题；但如果淘汰速度跟不上写入速度，字典长度会一直顶在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是不是合理，可能比看半天日志都快。缓存的难点从来不在“写入”和“命中”，而在“什么时候失效、占多少内存、并发怎么保护”——把这些想清楚，你写的任何一个小工具都能稳稳跑在生产环境里。