看到 Caveman 这个名字我第一个反应是这个项目够“硬核”。它不是什么花哨的框架也不是追逐热点的概念而是一个我最近从零写出来的极简内存缓存组件。说白了就是用一个 Python 类实现类似 Redis 的基础缓存能力键值存取、过期时间、LRU 淘汰、并发安全。叫 Caveman就是想表达“原始、直接、可靠”——像穴居人一样不需要复杂的工具也照样能把事情办了。这篇博文想跟你聊的就是这个“原始但能打”的缓存组件的完整设计思路、代码实现、生产落地过程中会遇到的各种坑以及我怎么一步步把它打磨到能上线的过程。无论你是后端开发、系统设计爱好者还是正在纠结“要不要为一个简单场景引入 Redis”的工程师这篇内容都值得你花几分钟看完。因为很多问题的答案并不是“越重越好”而是“刚刚好”。1. 项目思路与设计定位1.1 为什么叫 Caveman极简主义背后的取舍先说命名。我见过很多项目叫“Atlas”“Phoenix”“Skyfall”之类的听起来很宏大但实际代码里塞满了抽象层。 Caveman 这个名字是我故意取的因为我想做一个“退回到最原始状态但依然健壮”的组件。穴居人手里只有石头和火但他们照样活了几万年。我们的缓存不需要一堆 agent、cluster、持久化策略只需要解决掉“热点数据放哪里”这一个问题。这个取舍背后有真实场景有一次我在排查线上接口响应慢的问题发现一个配置数据每次请求都查数据库而这个配置实际每小时才变一次。当时项目里没有引入 Redis整个基础设施也不允许为一个几 KB 的配置去多维护一套高可用集群。我需要的只是“进程内存里放一份副本定时过期重读”。于是 Caveman 就诞生了一个纯 Python 实现、零外部依赖、开箱即用的内存缓存。当然极简不等于简陋。它需要有容量上限、过期时间、多线程读写安全、以及尽可能高的吞吐。这些核心要求一个都不能少但实现方式可以很朴素。很多团队一听到“缓存”就上 Redis却忽略了场景复杂度。Caveman 想证明的是90% 的技术需求根本用不着分布式单机内存缓存就已经绰绰有余。1.2 核心需求与适用场景写代码之前我先列了一组需求清单避免写着写着跑偏基本的 get、set、delete 操作支持 TTL 过期过期的 key 自动失效有容量上限超过上限时按 LRU 策略淘汰最久未使用的数据多线程环境下线程安全提供缓存命中率等统计信息方便排查问题零依赖装完 Python 就能跑。适用场景也很明确读多写少的配置数据、用户会话信息、临时令牌、热点榜单数据、以及一切“丢了重新拉取就行不会造成数据灾难”的内容。反过来如果你的数据一旦丢失会造成资金损失或者需要通过多节点同步来保证一致性那还是老老实实用专业缓存中间件Caveman 不背这个锅。这也引出了设计的第一条原则识别你的数据是否允许“偶尔不一致”。比如优惠券库存你绝对不能只靠本地缓存但用户昵称这种弱一致数据缓存 30 秒完全没问题。Caveman 的定位就是处理后者。1.3 技术选型数据结构层面的第一版方案技术上我选择了 Python 的 dict 作为底层存储。原因很简单dict 本身就是哈希表读操作平均 O(1)性能和实现成本都是最优解。但仅靠 dict 无法满足 LRU 淘汰需求因为 dict 不记录元素访问顺序。第一版我偷懒用了collections.OrderedDict通过move_to_end和popitem(lastFalse)实现 LRU。这个方案代码量很少大概 20 行就能跑通。但后来我放弃了因为OrderedDict虽然方便却把双向链表和哈希表的粘合逻辑封装得太死我想控制更底层的指针行为也方便后续扩展异步清理、更精确的过期扫描等功能。所以最终方案是用哈希表存 key 对应的节点引用用双向链表维护访问顺序。每次访问某个 key就把对应节点搬到链表头部需要淘汰时去掉链表尾部的节点。这套组合是 LRU 缓存的标准实现也是很多教科书里提到的做法。Caveman 整体架构也建立在这两个核心数据结构之上简单但高度可控。2. 核心数据结构与淘汰策略实现2.1 从哈希表说起O(1)读写的底层基础哈希表的原理这里简单提一下通过哈希函数把 key 映射到数组下标理想情况下查找时间是常数级。Python 内置 dict 已经做了大量优化包括冲突处理、扩容、哈希随机化等我们不用重复造轮子直接用就行。但在实现缓存时有一个细节需要注意dict 里存的 value 并不直接是用户数据而是链表节点对象。节点对象负责保存 key、value、过期时间、前后指针。这样当我们删除某个 key 时只需要从 dict 中删除同时把对应的节点从双向链表摘下来两段操作都是 O(1)。这里有个容易踩的坑如果 dict 保存的是节点对象那节点对象的哈希值不能依赖于 key 之外的字段。我们在代码里不会对节点做哈希而是对原始 key 做哈希所以没问题。但如果你在自定义__hash__时不小心包含了 value就可能导致同一个 key 的哈希值变化缓存查找直接失效。这类 bug 极其隐蔽排查起来非常痛苦。2.2 LRU 淘汰策略双向链表加哈希表怎么玩LRULeast Recently Used的核心思想是如果一份数据最近被访问过那未来被再次访问的概率也更高反过来很久没被访问过的数据就可以优先淘汰。实现 LRU 最经典的结构就是“哈希表 双向链表”。这里用双向链表而不是单向链表是因为在淘汰最久未使用节点时需要把尾部节点的前驱节点的 next 指针指向 None同时更新尾指针。如果只是单向链表你没法高效地从尾节点倒推前驱。双向链表多花了一个指针的存储开销但换来了 O(1) 的节点摘除操作。我把链表头尾分别设为哨兵节点dummy head 和 dummy tail避免大量“是否为空的”判断。代码看起来长一点但逻辑更统一。核心节点定义如下class _Node: __slots__ (key, value, expire_at, prev, next) def __init__(self, keyNone, valueNone, expire_atNone): self.key key self.value value self.expire_at expire_at self.prev None self.next None访问时把节点移动到链表头部def _move_to_head(self, node): self._remove_node(node) self._add_to_head(node)_remove_node做四件事把 node.prev 的 next 指向 node.next把 node.next 的 prev 指向 node.prev。_add_to_head则把 node 插到真正的头节点前面。哨兵节点的存在让这两个操作不需要判断边界。这里我不展开所有代码但会把核心逻辑写完整下一节给你一个可直接复制的实现。2.3 过期时间与惰性删除TTL 过期处理我采用了“惰性删除 主动清理”的组合策略。惰性删除指的是在 get 的时候检查当前时间是否超过 expire_at如果超过就直接当作不存在并删除。这个策略简单高效不需要额外线程但缺点是如果某些 key 过期后一直没人访问它们会一直占据内存直到达到容量上限触发淘汰。为了解决这个问题我加了一个后台“清理线程”每隔几秒扫描一次双向链表把过期的节点摘掉。扫描的起点是链表尾部因为尾部是最久未被访问的节点最可能是过期节点。只要遇到没过期的节点就停这里利用了“访问顺序”的特性来优化越靠近尾部的节点它的 expire_at 越可能较早。这里有个细节过期时间在节点创建时就会换算成绝对时间戳而不是保存相对秒数。为什么因为如果在 get 时用“当前时间 - 创建时间 ttl”来判断一旦系统时间被调整计算结果就不可靠。保存expire_at绝对时间判断逻辑直接变成now expire_at更简单也更稳定。如果运行环境有 NTP 时间同步也不容易出幺蛾子。3. 完整实现一个可运行的 Caveman Cache3.1 基础类结构与接口设计我把 Caveman Cache 设计成一个独立的类对外暴露尽量少的接口。我始终认为缓存组件的 API 越少越好因为调用方只需要知道“放进去”和“取出来”就够了。目前对外暴露的方法有set(key, value, ttlNone)写入缓存可覆盖默认过期时间get(key)读取缓存不存在或过期返回 Nonedelete(key)删除指定 keyclear()清空所有缓存stats()返回命中次数、未命中次数、当前容量等统计信息set_default_ttl(ttl)修改默认过期时间。内部类CavemanCache的初始化参数包括容量上限和默认 TTLclass CavemanCache: def __init__(self, capacity1024, default_ttl60): self.capacity capacity self.default_ttl default_ttl self._cache {} self._head _Node() self._tail _Node() self._head.next self._tail self._tail.prev self._head self._lock threading.RLock() self._hit_count 0 self._miss_count 0 self._expired_count 0构造函数里绕不开哨兵节点的初始化这是双向链表的基础设施。_cache字典负责存储 key 到节点的映射self._lock负责所有并发操作的互斥。之所以用RLock而不是Lock是因为允许同一个线程在持有锁的情况下再次进入后面实现原子操作时会更从容比如在 get 缺失回调的 set 中避免重复加锁。3.2 并发安全与锁粒度选择刚开始我考虑过用多把锁分别锁住字典和链表但这个粒度太细了容易死锁。比如线程 A 持有字典锁想获取链表锁线程 B 持有链表锁想获取字典锁就可能互相等待。与其做这种高风险优化不如直接上一把 RLock。因为 Caveman 的定位是进程内缓存单个操作的耗时都在微秒级一把锁的竞争开销完全可以接受。真正的优化重点是把耗时操作移出锁外。比如如果支持“缓存未命中自动加载数据”的回调函数那这个回调可能查数据库、调外部 API时间可能几十毫秒甚至几秒。你要是无脑在锁内执行整个缓存组件就变成了串行执行器性能直接崩掉。因此我设计了一个get_or_load方法先在锁内查缓存查不到就把加载逻辑放到锁外加载完成后再进行一次锁内写入。虽然比全程持锁多了一次锁入但并发能力提升了几十倍。生产并发下常见的一个坑是“重复加载”如果 100 个线程同时发现缓存不存在就会并发执行 100 次数据库查询。对于这种场景Caveman 里加了一个轻量级的 single-flight 机制用一个 dict 记录正在加载中的 key其他线程发现自己等就先睡一小会儿再去查缓存。这个机制不复杂但极其好用后面“缓存穿透”一节我会详细展开。3.3 关键 API 的实现细节set操作的流程是加锁、查字典。如果 key 已经存在说明节点已在链表中只需更新节点 value 和 expire_at再把它移到链表头部。如果 key 不存在需要新建节点插入字典同时挂到链表头部。如果此时容量超出上限就淘汰最后一个节点。get操作的流程更直白加锁、查字典。如果 key 不存在记一次 miss返回 None。如果 key 存在但已过期删除节点、记 expired返回 None。否则把节点移到链表头部更新 value 的临时热路径记一次 hit最后返回 value。这里的一个关键细节是移动到头部时如果直接在锁内操作需要保证删旧位置与插新位置不引发指针错乱。我的实现是先摘除节点再把节点插入头部整个过程都在锁内完成。因为_remove_node和_add_to_head都是 O(1) 指针操作锁内执行不会有性能负担。delete操作和clear操作很直观就不赘述了。需要特别注意的是清除链表时不要只清空字典一定要把头尾哨兵重新接上否则后续插入会 crash。这个低级错误我踩过测试用例都没发现后来是压测到高并发时偶发异常才排查出来。根因就是 clear 方法里我一开始没复位哨兵节点。下面是核心set和get的实现你可以直接抄走跑起来def set(self, key, value, ttlNone): expire_at time.time() (ttl if ttl is not None else self.default_ttl) with self._lock: node self._cache.get(key) if node is not None: node.value value node.expire_at expire_at self._move_to_head(node) return node _Node(key, value, expire_at) self._cache[key] node self._add_to_head(node) if len(self._cache) self.capacity: self._evict_tail() def get(self, key): now time.time() with self._lock: node self._cache.get(key) if node is None: self._miss_count 1 return None if node.expire_at and node.expire_at now: self._remove_node(node) del self._cache[key] self._expired_count 1 self._miss_count 1 return None self._hit_count 1 self._move_to_head(node) return node.value其中_evict_tail会从双向链表的尾部取出最久未使用节点如果该节点已经过期或者到达容量上限就删除它def _evict_tail(self): node self._tail.prev if node is not self._head: self._remove_node(node) del self._cache[node.key]这里注意_tail.prev是最后一个真实节点因为_tail是哨兵。所以每次淘汰的就是最久未访问节点完美契合 LRU 语义。4. 生产环境实战那些要命的坑4.1 缓存穿透、击穿、雪崩的应对先用最简单的话解释这三个问题。缓存穿透是指查询一个根本不存在的数据每次请求都绕过缓存直接打数据库。缓存击穿是指某个热点 key 过期的一瞬间大量请求同时涌入数据库。缓存雪崩是指大量 key 同时过期导致数据库压力瞬间飙升。Caveman 这层轻量级缓存也没法免俗我同样踩了这三个坑。穿透的应对方案很朴素缓存空值。当数据源查询结果为空时把 key 对应的 value 缓存成None的一个特殊标记对象TTL 设短一点比如 10 秒这样后续请求就不会再打数据库。需要注意的是直接缓存 Python 的 None 会和 Caveman 的“key 不存在”语义冲突因此我定义了一个内部哨兵类_EmptyValue用它作为 value 的包装。击穿的应对方案就是前面提到的 single-flight。这里我可以给你一个简化版思路用一个_loading_keys集合保存正在加载中的 key。线程 A 发现 key 不在缓存后先把 key 加入集合然后开始查询数据源线程 B 发现 key 在集合里就 sleep 几十毫秒然后重试 get。等到 A 加载完成并 set 进缓存B 第二次 get 时自然就能命中了。这个方案比引入分布式锁轻得多单机场景下完全够用。雪崩的应对方案是在业务层“打散过期时间”。比如统一设置 60 秒 TTL但实际写入时在 60 秒基础上加一个 0 到 30 秒之间的随机值。Caveman 支持在外部自行控制 ttl我就在调用层做了这个小改动。这一招成本极低但对数据库的保护效果立竿见影。4.2 序列化格式与内存占用Caveman 默认只支持在进程内存储 Python 对象但很多项目需要把缓存内容发给其他服务使用这就涉及序列化。我在后续扩展中选择使用 JSON 格式原因是对外可读、排障方便。但它有两个硬伤一是无法序列化 datetime、bytes 等类型需要额外转换二是序列化和反序列化的 CPU 开销比 Pickle 高。如果只是进程内缓存直接持有对象引用反而更高效。内存占用方面我遇到过一个问题Caveman 的节点对象如果使用常规的__dict__存储属性会额外消耗不少内存。一个带五个属性的节点__dict__可能占 200 字节以上。我后来给节点类加上了__slots__让每个属性变成固定槽内存占用直接降了三分之一。对于缓存容量 10 万的场景这省下来的内存肉眼可见。还有一个容易忽略的点如果缓存的值本身是可变对象而且调用方在拿到引用后修改了它缓存里的数据也会跟着变。这个问题很难从 Caveman 层面完全规避我的建议是在写入时复制一份或者在业务上约定“缓存内的对象只读”。别小看这个约定很多线上脏数据问题就是这么来的。4.3 可观测性metrics、日志与故障排查一开始 Caveman 只暴露了stats()但线上排查时局限性很大。因为统计数据需要知道“命中率正在下降”“过期清理线程卡住”这些趋势而不是只能看到当前瞬间的值。我给加了一个_record_stats机制每次读写时原子更新计数并在stats()里返回快照。同时我加了事件日志把 key 数量变化、淘汰数量、过期清理数量按分钟维度打印出来方便对齐业务高峰期。日志输出我用的是标准库logging日志级别区分得很开正常信息用 DEBUG淘汰和过期清理用 INFO异常用 WARNING。生产环境只开 WARNING开发环境开 DEBUG。这里有个小心得日志里不要记录完整的 value尤其是会话 token 或用户资料否则日志文件就是数据库级别的信息泄露。打日志时只记录 key 和长度。还有一个排查坑如果你在 set 之前没有检查 value 是否为 None那么后续 get 的返回和“key 不存在”很难区分。我最终决定Caveman 禁止用户把 None 作为 value 传入。如果确实需要缓存 None就用_EmptyValue包装。这个约定从根本上杜绝了歧义值得任何缓存组件都默认遵守。5. 测试与性能验证5.1 单元测试设计覆盖断点不是越多越好我为 Caveman 设计的单元测试主要是围绕三种行为基本读写、过期、淘汰、并发。基本读写测试很简单就是 set 之后 get验证返回原值delete 之后 get验证返回 None。但需要注意的是TTL 测试不能真的 sleep否则测试会又慢又不稳定。我的做法是把time.time替换成可拨动的 mock直接推进时钟到 61 秒来触发过期测试速度快也不会因为机器卡顿出现偶发失败。LRU 淘汰测试要重点验证“容量满时插入新 key 会淘汰最久未使用的 key”。这个测试很容易写错因为连续访问顺序不同导致结果不同。我建议用这个序列做断言cache CavemanCache(capacity2) cache.set(a, 1) cache.set(b, 2) cache.get(a) # a 变为最近使用 cache.set(c, 3) # 应该淘汰 b assert cache.get(b) is None assert cache.get(a) 1 assert cache.get(c) 3并发测试可以用 ThreadPoolExecutor 起 8 个线程同时对同一个 key 做 1000 次读和写最后断言缓存数据一致且没有异常。这种测试不一定能暴露所有问题但至少能验证锁的基本互斥性。真要说高并发下锁设计有没有问题还得靠压测。5.2 压测结果与参数调整我压测环境是 4 核 8G 的云服务器Python 3.10。单线程顺序读写 100 万次Caveman 大约耗时 1.8 秒折算吞吐 55 万 QPS 左右。8 线程并发读写时因为 GIL 和 RLock 竞争吞吐会掉到 20 万 QPS 左右但在单机缓存场景下已经非常够用。压测中我发现容量参数对性能有微妙影响容量太小会导致频繁淘汰容量太大则内存占用偏高。比如容量设为 1024热点数据量 2000 时缓存命中率只有 60%性能提升有限。后来我把容量调到热点数据的 1.5 倍配合过期时间打散命中率提升到 98%。这里一个重要经验缓存容量的基准不是总的 key 数而是“单位时间内活跃访问的 key 数”。我还专门压过过期清理线程扫描 10 万节点时单轮耗时大概 80 毫秒。如果每 10 秒扫一轮对主链路的影响可以忽略。但如果你把清理频率设到 100 毫秒一次CPU 就会被白白吃掉。所以默认清理频率 10 秒一轮已经足够。5.3 和其他成熟方案的对比Redis 不是唯一答案最后对比一下 Caveman 和主流方案。Redis 功能丰富、持久化、高可用、可跨进程是分布式场景的王者。但对于单机 Python 服务引入 Redis 意味着新增运维依赖、网络开销、序列化成本。Caveman 零依赖直接共享进程内存性能更高。另一个常见的 Python 库是cachetools它提供了多种缓存策略包括 LRU、TTL、LFU代码质量和生态都很好。如果你不想造轮子直接用 cachetools 完全合理。Caveman 的价值在于它是你自己可控的代码你能完全理解它的每一行遇到特殊需求可以随时改。比如我想加 single-flight、异步清理、日志统计cachetools 要绕很多弯子Caveman 改起来就非常顺手。下面这个表格可以帮你快速决策方案依赖部署复杂度适用场景性能级别Caveman 内存缓存无极低单进程应用、轻量缓存单机高吞吐cachetools轻量低需要快速接入多种策略较高Redis重高多进程/多服务共享缓存网络延迟需序列化如果你只是想给一个简单接口加缓存Caveman 这种轻量方案是第一优先如果你的服务未来一定会扩展到多实例那从一开始就考虑 Redis 更明智。但就算用了 RedisCaveman 也能作为本地一级缓存挡掉大部分热点请求减少对 Redis 的压力。我个人在实际操作中的体会是这类“自己造一个极简组件”的项目最宝贵的收获不是代码量而是对缓存原理有了肌肉记忆式的理解。后来我再排查线上遗漏缓存问题时一眼就能猜出大概是淘汰策略、过期时间还是并发锁的问题。如果你现在想练手建议从 Caveman 起步然后把 single-flight、清理线程、持久化逐步加进去每次只增加一个能力确保每步都可测试、可回滚。这样扩展出来的组件远比一上来就设计一个“万能缓存框架”可靠得多。最后一个小技巧给组件写一个非常简单的基准测试脚本每次改动后跑一遍性能负优化会第一时间警告你这个习惯能帮你少踩很多雷。
看到 Caveman 这个名字我第一个反应是这个项目够“硬核”。它不是什么花哨的框架也不是追逐热点的概念而是一个我最近从零写出来的极简内存缓存组件。说白了就是用一个 Python 类实现类似 Redis 的基础缓存能力键值存取、过期时间、LRU 淘汰、并发安全。叫 Caveman就是想表达“原始、直接、可靠”——像穴居人一样不需要复杂的工具也照样能把事情办了。这篇博文想跟你聊的就是这个“原始但能打”的缓存组件的完整设计思路、代码实现、生产落地过程中会遇到的各种坑以及我怎么一步步把它打磨到能上线的过程。无论你是后端开发、系统设计爱好者还是正在纠结“要不要为一个简单场景引入 Redis”的工程师这篇内容都值得你花几分钟看完。因为很多问题的答案并不是“越重越好”而是“刚刚好”。1. 项目思路与设计定位1.1 为什么叫 Caveman极简主义背后的取舍先说命名。我见过很多项目叫“Atlas”“Phoenix”“Skyfall”之类的听起来很宏大但实际代码里塞满了抽象层。 Caveman 这个名字是我故意取的因为我想做一个“退回到最原始状态但依然健壮”的组件。穴居人手里只有石头和火但他们照样活了几万年。我们的缓存不需要一堆 agent、cluster、持久化策略只需要解决掉“热点数据放哪里”这一个问题。这个取舍背后有真实场景有一次我在排查线上接口响应慢的问题发现一个配置数据每次请求都查数据库而这个配置实际每小时才变一次。当时项目里没有引入 Redis整个基础设施也不允许为一个几 KB 的配置去多维护一套高可用集群。我需要的只是“进程内存里放一份副本定时过期重读”。于是 Caveman 就诞生了一个纯 Python 实现、零外部依赖、开箱即用的内存缓存。当然极简不等于简陋。它需要有容量上限、过期时间、多线程读写安全、以及尽可能高的吞吐。这些核心要求一个都不能少但实现方式可以很朴素。很多团队一听到“缓存”就上 Redis却忽略了场景复杂度。Caveman 想证明的是90% 的技术需求根本用不着分布式单机内存缓存就已经绰绰有余。1.2 核心需求与适用场景写代码之前我先列了一组需求清单避免写着写着跑偏基本的 get、set、delete 操作支持 TTL 过期过期的 key 自动失效有容量上限超过上限时按 LRU 策略淘汰最久未使用的数据多线程环境下线程安全提供缓存命中率等统计信息方便排查问题零依赖装完 Python 就能跑。适用场景也很明确读多写少的配置数据、用户会话信息、临时令牌、热点榜单数据、以及一切“丢了重新拉取就行不会造成数据灾难”的内容。反过来如果你的数据一旦丢失会造成资金损失或者需要通过多节点同步来保证一致性那还是老老实实用专业缓存中间件Caveman 不背这个锅。这也引出了设计的第一条原则识别你的数据是否允许“偶尔不一致”。比如优惠券库存你绝对不能只靠本地缓存但用户昵称这种弱一致数据缓存 30 秒完全没问题。Caveman 的定位就是处理后者。1.3 技术选型数据结构层面的第一版方案技术上我选择了 Python 的 dict 作为底层存储。原因很简单dict 本身就是哈希表读操作平均 O(1)性能和实现成本都是最优解。但仅靠 dict 无法满足 LRU 淘汰需求因为 dict 不记录元素访问顺序。第一版我偷懒用了collections.OrderedDict通过move_to_end和popitem(lastFalse)实现 LRU。这个方案代码量很少大概 20 行就能跑通。但后来我放弃了因为OrderedDict虽然方便却把双向链表和哈希表的粘合逻辑封装得太死我想控制更底层的指针行为也方便后续扩展异步清理、更精确的过期扫描等功能。所以最终方案是用哈希表存 key 对应的节点引用用双向链表维护访问顺序。每次访问某个 key就把对应节点搬到链表头部需要淘汰时去掉链表尾部的节点。这套组合是 LRU 缓存的标准实现也是很多教科书里提到的做法。Caveman 整体架构也建立在这两个核心数据结构之上简单但高度可控。2. 核心数据结构与淘汰策略实现2.1 从哈希表说起O(1)读写的底层基础哈希表的原理这里简单提一下通过哈希函数把 key 映射到数组下标理想情况下查找时间是常数级。Python 内置 dict 已经做了大量优化包括冲突处理、扩容、哈希随机化等我们不用重复造轮子直接用就行。但在实现缓存时有一个细节需要注意dict 里存的 value 并不直接是用户数据而是链表节点对象。节点对象负责保存 key、value、过期时间、前后指针。这样当我们删除某个 key 时只需要从 dict 中删除同时把对应的节点从双向链表摘下来两段操作都是 O(1)。这里有个容易踩的坑如果 dict 保存的是节点对象那节点对象的哈希值不能依赖于 key 之外的字段。我们在代码里不会对节点做哈希而是对原始 key 做哈希所以没问题。但如果你在自定义__hash__时不小心包含了 value就可能导致同一个 key 的哈希值变化缓存查找直接失效。这类 bug 极其隐蔽排查起来非常痛苦。2.2 LRU 淘汰策略双向链表加哈希表怎么玩LRULeast Recently Used的核心思想是如果一份数据最近被访问过那未来被再次访问的概率也更高反过来很久没被访问过的数据就可以优先淘汰。实现 LRU 最经典的结构就是“哈希表 双向链表”。这里用双向链表而不是单向链表是因为在淘汰最久未使用节点时需要把尾部节点的前驱节点的 next 指针指向 None同时更新尾指针。如果只是单向链表你没法高效地从尾节点倒推前驱。双向链表多花了一个指针的存储开销但换来了 O(1) 的节点摘除操作。我把链表头尾分别设为哨兵节点dummy head 和 dummy tail避免大量“是否为空的”判断。代码看起来长一点但逻辑更统一。核心节点定义如下class _Node: __slots__ (key, value, expire_at, prev, next) def __init__(self, keyNone, valueNone, expire_atNone): self.key key self.value value self.expire_at expire_at self.prev None self.next None访问时把节点移动到链表头部def _move_to_head(self, node): self._remove_node(node) self._add_to_head(node)_remove_node做四件事把 node.prev 的 next 指向 node.next把 node.next 的 prev 指向 node.prev。_add_to_head则把 node 插到真正的头节点前面。哨兵节点的存在让这两个操作不需要判断边界。这里我不展开所有代码但会把核心逻辑写完整下一节给你一个可直接复制的实现。2.3 过期时间与惰性删除TTL 过期处理我采用了“惰性删除 主动清理”的组合策略。惰性删除指的是在 get 的时候检查当前时间是否超过 expire_at如果超过就直接当作不存在并删除。这个策略简单高效不需要额外线程但缺点是如果某些 key 过期后一直没人访问它们会一直占据内存直到达到容量上限触发淘汰。为了解决这个问题我加了一个后台“清理线程”每隔几秒扫描一次双向链表把过期的节点摘掉。扫描的起点是链表尾部因为尾部是最久未被访问的节点最可能是过期节点。只要遇到没过期的节点就停这里利用了“访问顺序”的特性来优化越靠近尾部的节点它的 expire_at 越可能较早。这里有个细节过期时间在节点创建时就会换算成绝对时间戳而不是保存相对秒数。为什么因为如果在 get 时用“当前时间 - 创建时间 ttl”来判断一旦系统时间被调整计算结果就不可靠。保存expire_at绝对时间判断逻辑直接变成now expire_at更简单也更稳定。如果运行环境有 NTP 时间同步也不容易出幺蛾子。3. 完整实现一个可运行的 Caveman Cache3.1 基础类结构与接口设计我把 Caveman Cache 设计成一个独立的类对外暴露尽量少的接口。我始终认为缓存组件的 API 越少越好因为调用方只需要知道“放进去”和“取出来”就够了。目前对外暴露的方法有set(key, value, ttlNone)写入缓存可覆盖默认过期时间get(key)读取缓存不存在或过期返回 Nonedelete(key)删除指定 keyclear()清空所有缓存stats()返回命中次数、未命中次数、当前容量等统计信息set_default_ttl(ttl)修改默认过期时间。内部类CavemanCache的初始化参数包括容量上限和默认 TTLclass CavemanCache: def __init__(self, capacity1024, default_ttl60): self.capacity capacity self.default_ttl default_ttl self._cache {} self._head _Node() self._tail _Node() self._head.next self._tail self._tail.prev self._head self._lock threading.RLock() self._hit_count 0 self._miss_count 0 self._expired_count 0构造函数里绕不开哨兵节点的初始化这是双向链表的基础设施。_cache字典负责存储 key 到节点的映射self._lock负责所有并发操作的互斥。之所以用RLock而不是Lock是因为允许同一个线程在持有锁的情况下再次进入后面实现原子操作时会更从容比如在 get 缺失回调的 set 中避免重复加锁。3.2 并发安全与锁粒度选择刚开始我考虑过用多把锁分别锁住字典和链表但这个粒度太细了容易死锁。比如线程 A 持有字典锁想获取链表锁线程 B 持有链表锁想获取字典锁就可能互相等待。与其做这种高风险优化不如直接上一把 RLock。因为 Caveman 的定位是进程内缓存单个操作的耗时都在微秒级一把锁的竞争开销完全可以接受。真正的优化重点是把耗时操作移出锁外。比如如果支持“缓存未命中自动加载数据”的回调函数那这个回调可能查数据库、调外部 API时间可能几十毫秒甚至几秒。你要是无脑在锁内执行整个缓存组件就变成了串行执行器性能直接崩掉。因此我设计了一个get_or_load方法先在锁内查缓存查不到就把加载逻辑放到锁外加载完成后再进行一次锁内写入。虽然比全程持锁多了一次锁入但并发能力提升了几十倍。生产并发下常见的一个坑是“重复加载”如果 100 个线程同时发现缓存不存在就会并发执行 100 次数据库查询。对于这种场景Caveman 里加了一个轻量级的 single-flight 机制用一个 dict 记录正在加载中的 key其他线程发现自己等就先睡一小会儿再去查缓存。这个机制不复杂但极其好用后面“缓存穿透”一节我会详细展开。3.3 关键 API 的实现细节set操作的流程是加锁、查字典。如果 key 已经存在说明节点已在链表中只需更新节点 value 和 expire_at再把它移到链表头部。如果 key 不存在需要新建节点插入字典同时挂到链表头部。如果此时容量超出上限就淘汰最后一个节点。get操作的流程更直白加锁、查字典。如果 key 不存在记一次 miss返回 None。如果 key 存在但已过期删除节点、记 expired返回 None。否则把节点移到链表头部更新 value 的临时热路径记一次 hit最后返回 value。这里的一个关键细节是移动到头部时如果直接在锁内操作需要保证删旧位置与插新位置不引发指针错乱。我的实现是先摘除节点再把节点插入头部整个过程都在锁内完成。因为_remove_node和_add_to_head都是 O(1) 指针操作锁内执行不会有性能负担。delete操作和clear操作很直观就不赘述了。需要特别注意的是清除链表时不要只清空字典一定要把头尾哨兵重新接上否则后续插入会 crash。这个低级错误我踩过测试用例都没发现后来是压测到高并发时偶发异常才排查出来。根因就是 clear 方法里我一开始没复位哨兵节点。下面是核心set和get的实现你可以直接抄走跑起来def set(self, key, value, ttlNone): expire_at time.time() (ttl if ttl is not None else self.default_ttl) with self._lock: node self._cache.get(key) if node is not None: node.value value node.expire_at expire_at self._move_to_head(node) return node _Node(key, value, expire_at) self._cache[key] node self._add_to_head(node) if len(self._cache) self.capacity: self._evict_tail() def get(self, key): now time.time() with self._lock: node self._cache.get(key) if node is None: self._miss_count 1 return None if node.expire_at and node.expire_at now: self._remove_node(node) del self._cache[key] self._expired_count 1 self._miss_count 1 return None self._hit_count 1 self._move_to_head(node) return node.value其中_evict_tail会从双向链表的尾部取出最久未使用节点如果该节点已经过期或者到达容量上限就删除它def _evict_tail(self): node self._tail.prev if node is not self._head: self._remove_node(node) del self._cache[node.key]这里注意_tail.prev是最后一个真实节点因为_tail是哨兵。所以每次淘汰的就是最久未访问节点完美契合 LRU 语义。4. 生产环境实战那些要命的坑4.1 缓存穿透、击穿、雪崩的应对先用最简单的话解释这三个问题。缓存穿透是指查询一个根本不存在的数据每次请求都绕过缓存直接打数据库。缓存击穿是指某个热点 key 过期的一瞬间大量请求同时涌入数据库。缓存雪崩是指大量 key 同时过期导致数据库压力瞬间飙升。Caveman 这层轻量级缓存也没法免俗我同样踩了这三个坑。穿透的应对方案很朴素缓存空值。当数据源查询结果为空时把 key 对应的 value 缓存成None的一个特殊标记对象TTL 设短一点比如 10 秒这样后续请求就不会再打数据库。需要注意的是直接缓存 Python 的 None 会和 Caveman 的“key 不存在”语义冲突因此我定义了一个内部哨兵类_EmptyValue用它作为 value 的包装。击穿的应对方案就是前面提到的 single-flight。这里我可以给你一个简化版思路用一个_loading_keys集合保存正在加载中的 key。线程 A 发现 key 不在缓存后先把 key 加入集合然后开始查询数据源线程 B 发现 key 在集合里就 sleep 几十毫秒然后重试 get。等到 A 加载完成并 set 进缓存B 第二次 get 时自然就能命中了。这个方案比引入分布式锁轻得多单机场景下完全够用。雪崩的应对方案是在业务层“打散过期时间”。比如统一设置 60 秒 TTL但实际写入时在 60 秒基础上加一个 0 到 30 秒之间的随机值。Caveman 支持在外部自行控制 ttl我就在调用层做了这个小改动。这一招成本极低但对数据库的保护效果立竿见影。4.2 序列化格式与内存占用Caveman 默认只支持在进程内存储 Python 对象但很多项目需要把缓存内容发给其他服务使用这就涉及序列化。我在后续扩展中选择使用 JSON 格式原因是对外可读、排障方便。但它有两个硬伤一是无法序列化 datetime、bytes 等类型需要额外转换二是序列化和反序列化的 CPU 开销比 Pickle 高。如果只是进程内缓存直接持有对象引用反而更高效。内存占用方面我遇到过一个问题Caveman 的节点对象如果使用常规的__dict__存储属性会额外消耗不少内存。一个带五个属性的节点__dict__可能占 200 字节以上。我后来给节点类加上了__slots__让每个属性变成固定槽内存占用直接降了三分之一。对于缓存容量 10 万的场景这省下来的内存肉眼可见。还有一个容易忽略的点如果缓存的值本身是可变对象而且调用方在拿到引用后修改了它缓存里的数据也会跟着变。这个问题很难从 Caveman 层面完全规避我的建议是在写入时复制一份或者在业务上约定“缓存内的对象只读”。别小看这个约定很多线上脏数据问题就是这么来的。4.3 可观测性metrics、日志与故障排查一开始 Caveman 只暴露了stats()但线上排查时局限性很大。因为统计数据需要知道“命中率正在下降”“过期清理线程卡住”这些趋势而不是只能看到当前瞬间的值。我给加了一个_record_stats机制每次读写时原子更新计数并在stats()里返回快照。同时我加了事件日志把 key 数量变化、淘汰数量、过期清理数量按分钟维度打印出来方便对齐业务高峰期。日志输出我用的是标准库logging日志级别区分得很开正常信息用 DEBUG淘汰和过期清理用 INFO异常用 WARNING。生产环境只开 WARNING开发环境开 DEBUG。这里有个小心得日志里不要记录完整的 value尤其是会话 token 或用户资料否则日志文件就是数据库级别的信息泄露。打日志时只记录 key 和长度。还有一个排查坑如果你在 set 之前没有检查 value 是否为 None那么后续 get 的返回和“key 不存在”很难区分。我最终决定Caveman 禁止用户把 None 作为 value 传入。如果确实需要缓存 None就用_EmptyValue包装。这个约定从根本上杜绝了歧义值得任何缓存组件都默认遵守。5. 测试与性能验证5.1 单元测试设计覆盖断点不是越多越好我为 Caveman 设计的单元测试主要是围绕三种行为基本读写、过期、淘汰、并发。基本读写测试很简单就是 set 之后 get验证返回原值delete 之后 get验证返回 None。但需要注意的是TTL 测试不能真的 sleep否则测试会又慢又不稳定。我的做法是把time.time替换成可拨动的 mock直接推进时钟到 61 秒来触发过期测试速度快也不会因为机器卡顿出现偶发失败。LRU 淘汰测试要重点验证“容量满时插入新 key 会淘汰最久未使用的 key”。这个测试很容易写错因为连续访问顺序不同导致结果不同。我建议用这个序列做断言cache CavemanCache(capacity2) cache.set(a, 1) cache.set(b, 2) cache.get(a) # a 变为最近使用 cache.set(c, 3) # 应该淘汰 b assert cache.get(b) is None assert cache.get(a) 1 assert cache.get(c) 3并发测试可以用 ThreadPoolExecutor 起 8 个线程同时对同一个 key 做 1000 次读和写最后断言缓存数据一致且没有异常。这种测试不一定能暴露所有问题但至少能验证锁的基本互斥性。真要说高并发下锁设计有没有问题还得靠压测。5.2 压测结果与参数调整我压测环境是 4 核 8G 的云服务器Python 3.10。单线程顺序读写 100 万次Caveman 大约耗时 1.8 秒折算吞吐 55 万 QPS 左右。8 线程并发读写时因为 GIL 和 RLock 竞争吞吐会掉到 20 万 QPS 左右但在单机缓存场景下已经非常够用。压测中我发现容量参数对性能有微妙影响容量太小会导致频繁淘汰容量太大则内存占用偏高。比如容量设为 1024热点数据量 2000 时缓存命中率只有 60%性能提升有限。后来我把容量调到热点数据的 1.5 倍配合过期时间打散命中率提升到 98%。这里一个重要经验缓存容量的基准不是总的 key 数而是“单位时间内活跃访问的 key 数”。我还专门压过过期清理线程扫描 10 万节点时单轮耗时大概 80 毫秒。如果每 10 秒扫一轮对主链路的影响可以忽略。但如果你把清理频率设到 100 毫秒一次CPU 就会被白白吃掉。所以默认清理频率 10 秒一轮已经足够。5.3 和其他成熟方案的对比Redis 不是唯一答案最后对比一下 Caveman 和主流方案。Redis 功能丰富、持久化、高可用、可跨进程是分布式场景的王者。但对于单机 Python 服务引入 Redis 意味着新增运维依赖、网络开销、序列化成本。Caveman 零依赖直接共享进程内存性能更高。另一个常见的 Python 库是cachetools它提供了多种缓存策略包括 LRU、TTL、LFU代码质量和生态都很好。如果你不想造轮子直接用 cachetools 完全合理。Caveman 的价值在于它是你自己可控的代码你能完全理解它的每一行遇到特殊需求可以随时改。比如我想加 single-flight、异步清理、日志统计cachetools 要绕很多弯子Caveman 改起来就非常顺手。下面这个表格可以帮你快速决策方案依赖部署复杂度适用场景性能级别Caveman 内存缓存无极低单进程应用、轻量缓存单机高吞吐cachetools轻量低需要快速接入多种策略较高Redis重高多进程/多服务共享缓存网络延迟需序列化如果你只是想给一个简单接口加缓存Caveman 这种轻量方案是第一优先如果你的服务未来一定会扩展到多实例那从一开始就考虑 Redis 更明智。但就算用了 RedisCaveman 也能作为本地一级缓存挡掉大部分热点请求减少对 Redis 的压力。我个人在实际操作中的体会是这类“自己造一个极简组件”的项目最宝贵的收获不是代码量而是对缓存原理有了肌肉记忆式的理解。后来我再排查线上遗漏缓存问题时一眼就能猜出大概是淘汰策略、过期时间还是并发锁的问题。如果你现在想练手建议从 Caveman 起步然后把 single-flight、清理线程、持久化逐步加进去每次只增加一个能力确保每步都可测试、可回滚。这样扩展出来的组件远比一上来就设计一个“万能缓存框架”可靠得多。最后一个小技巧给组件写一个非常简单的基准测试脚本每次改动后跑一遍性能负优化会第一时间警告你这个习惯能帮你少踩很多雷。
