本地缓存的需求通常长这样:热点配置只有几百条,读多写少,希望固定内存上限,最久没被访问的先淘汰。Redis 能解决,但为几百条数据多一次网络往返不划算;functools.lru_cache 又只适合「函数入参到返回值」的纯函数场景。
自己写一个 LRU,核心只有两件事:哈希表负责 O(1) 查找,双向链表负责 O(1) 调整顺序。数组做不到,因为把中间元素移到头部需要搬移;单链表也不行,删除一个节点需要找前驱。
结构:两个哨兵省掉一半边界判断#
链表头是最近访问,尾部是最久未访问。插入、删除都围绕头尾进行,所以放两个哨兵节点 head 和 tail,让空链表、只有一个节点、删除尾节点这几种情况共用一套指针操作。
head <-> [最近] <-> [ ... ] <-> [最久] <-> tail
get/put 命中时移到 head 前
容量满时删除 tail.prev
每个节点同时存 key 和 value。只存 value 不够,因为淘汰尾节点时还要用 key 去哈希表里删索引。
Python 版:完整可运行#
from __future__ import annotations
from typing import Dict, Generic, Optional, TypeVar
K = TypeVar("K")
V = TypeVar("V")
class _Node(Generic[K, V]):
__slots__ = ("key", "value", "prev", "next")
def __init__(self, key: Optional[K] = None, value: Optional[V] = None):
self.key = key
self.value = value
self.prev: Optional[_Node[K, V]] = None
self.next: Optional[_Node[K, V]] = None
class LRUCache(Generic[K, V]):
def __init__(self, capacity: int):
if capacity < 1:
raise ValueError("capacity must be >= 1")
self.capacity = capacity
self._map: Dict[K, _Node[K, V]] = {}
self._head: _Node[K, V] = _Node()
self._tail: _Node[K, V] = _Node()
self._head.next = self._tail
self._tail.prev = self._head
def _unlink(self, node: _Node[K, V]) -> None:
assert node.prev and node.next
node.prev.next = node.next
node.next.prev = node.prev
def _push_front(self, node: _Node[K, V]) -> None:
node.next = self._head.next
node.prev = self._head
assert self._head.next
self._head.next.prev = node
self._head.next = node
def get(self, key: K) -> Optional[V]:
node = self._map.get(key)
if node is None:
return None
self._unlink(node)
self._push_front(node)
return node.value
def put(self, key: K, value: V) -> None:
node = self._map.get(key)
if node is not None:
node.value = value
self._unlink(node)
self._push_front(node)
return
if len(self._map) >= self.capacity:
victim = self._tail.prev
assert victim and victim is not self._head and victim.key is not None
self._unlink(victim)
del self._map[victim.key]
node = _Node(key, value)
self._map[key] = node
self._push_front(node)
def __len__(self) -> int:
return len(self._map)
用法和边界:
cache = LRUCache[str, int](2)
cache.put("a", 1)
cache.put("b", 2)
cache.get("a") # 命中,a 变成最近使用
cache.put("c", 3) # 淘汰 b
assert cache.get("b") is None
assert cache.get("a") == 1
这里有一个设计取舍:get 用 None 表示未命中,因此缓存里不能存 None。如果业务需要存空值,把返回值改成 tuple[bool, V | None],或者定义一个模块级 MISSING = object() 作为默认值。不要用 key in map 再 get,那是两次查找。
TypeScript 版:利用 Map 的插入顺序#
JavaScript 的 Map 会保留键的插入顺序,delete 后再 set 会把键移到末尾。把「末尾」当作最近使用,代码可以短很多:
export class LRUCache<K, V> {
private readonly map = new Map<K, V>();
constructor(private readonly capacity: number) {
if (!Number.isInteger(capacity) || capacity < 1) {
throw new Error("capacity must be a positive integer");
}
}
get(key: K): V | undefined {
if (!this.map.has(key)) return undefined;
const value = this.map.get(key) as V;
this.map.delete(key);
this.map.set(key, value);
return value;
}
put(key: K, value: V): void {
if (this.map.has(key)) {
this.map.delete(key);
} else if (this.map.size >= this.capacity) {
const oldest = this.map.keys().next().value as K | undefined;
if (oldest !== undefined) this.map.delete(oldest);
}
this.map.set(key, value);
}
get size(): number {
return this.map.size;
}
}
TypeScript 版本同样有「不能缓存 undefined」的限制。另一个坑是 oldest !== undefined:如果业务真的用 undefined 作为 key,这里会误判。生产代码要么禁止 undefined key,要么改用显式的双向链表版本。
并发:O(1) 不等于线程安全#
Python 有 GIL,但 get 里的「查表 → 摘链 → 插头」是三段操作,线程切换可能发生在任意字节码之间。多线程环境下要加锁:
import threading
class ThreadSafeLRU(LRUCache[K, V]):
def __init__(self, capacity: int):
super().__init__(capacity)
self._lock = threading.Lock()
def get(self, key: K) -> Optional[V]:
with self._lock:
return super().get(key)
def put(self, key: K, value: V) -> None:
with self._lock:
super().put(key, value)
锁的粒度要覆盖整个操作,不能只锁哈希表。更好的做法是把「查缓存」和「回源计算」分开:锁内只做查/写,回源在锁外;同时用 single-flight 合并同一个 key 的并发回源,否则缓存未命中时会有 N 个请求一起打到数据库。
TypeScript 的单线程只保证同步代码不交错,async 函数在 await 处仍会让出。不要把「先 get 判断,再 await 回源,再 put」当成原子操作;同一 key 的并发请求需要 Promise 去重。
容量怎么估,不要拍脑袋#
LRU 的容量不是越大越好。每条记录除了 value 本身,还有哈希表槽位、节点对象、key 的引用,Python 下一条小对象往往要几百字节。缓存十万条配置,可能先吃掉几十 MB 内存,换来的命中率却只比一万条高几个百分点。
先用 tracemalloc 或 sys.getsizeof 做一次粗估,再按「内存预算 ÷ 单条平均开销」定上限;上线后观察命中率和淘汰速率,命中率不再明显上升时就该停止加容量。TypeScript 的 Map 同理,V8 的对象开销和字符串驻留都会影响实际占用。
什么时候别自己写#
| 场景 | 建议 |
|---|---|
| 纯函数结果缓存 | functools.lru_cache(maxsize=...) |
| 需要 TTL、统计、多种淘汰策略 | cachetools 或 Caffeine |
| 多进程 / 多实例共享 | Redis,别用进程内缓存 |
| 缓存对象很大、更新频繁 | 先算内存上限,再决定容量 |
| 需要持久化 | 数据库或带持久化的缓存服务 |
自己写的价值在于可控:容量、淘汰顺序、空值语义、并发策略都能按业务调整。代价是测试要自己补,尤其是容量为 1、重复 put、命中后淘汰顺序这三组用例。
结论#
- LRU 的本质是哈希表 + 双向链表,哨兵节点能把边界判断收敛成一套指针操作。
- Python 版用
__slots__降低节点开销;TypeScript 版可以用Map的插入顺序写出更短的实现。 None/undefined与「未命中」的语义要提前定义,不要让调用方猜。- 并发下必须锁住完整操作;缓存未命中时的回源,用 single-flight 防止击穿。
- 进程内缓存不解决多实例一致性问题,跨进程共享请直接用 Redis。
完