线上有个商品详情接口，单机 QPS 6000，DB 顶不住，第一反应是加本地缓存。很多人的第一版长这样：
var cache = make(map[string][]byte)
var mu sync.Mutex
func Get(key string) []byte {
mu.Lock()
defer mu.Unlock()
return cache[key]
}
func Set(key string, v []byte) {
mu.Lock()
defer mu.Unlock()
cache[key] = v
}
上线三天，容器被 OOMKilled。key 空间是「商品 ID × 站点 × 语言」，一天能碰到几百万个组合，value 序列化后 10~30KB，内存只增不减。
有人改成「超过 10 万条就整个清空」，结果更糟：每隔几十分钟来一次全量击穿，DB 瞬间被打到 3 万 QPS。需要的是「有上限 + 淘汰时挑最不该留的那个」，这就是 LRU 要解决的问题。下面手写一遍，然后把踩到的坑一个个列出来。
为什么是哈希表 + 双向链表
LRU 要对两个操作都保持 O(1)：
- 按 key 找到值 —— 只有哈希表能做到。
- 把刚访问的节点挪到「最新」端，从「最旧」端淘汰 —— 要能在 O(1) 时间把一个节点从链表中间摘出来，并且知道前后是谁。
数组不行，中间删除要搬移元素，O(n)。单链表不行，摘除一个节点需要它的前驱，从头找是 O(n)。双向链表可以，节点自己就存了 prev/next。
关键的一点，也是最多人写错的一点：哈希表里存的必须是指向链表节点的指针，不能是值本身。map[K]V 只能拿到值，拿不到节点，也就没法 O(1) 摘除。
结构大概长这样，head 侧是最近使用，tail 侧是最久未使用：
head <-> [A] <-> [B] <-> [C] <-> tail
↑ 最新 ↑ 最旧，满了就淘汰它
map: "A" -> *nodeA, "B" -> *nodeB, "C" -> *nodeC
用两个哨兵节点（head/tail）可以省掉所有「链表为空」「插到头部」「删除尾节点」的空指针判断，这是双向链表实现里最省事的一个技巧。
手写实现：Go 泛型版
完整代码，可以直接跑，链表用哨兵，map 存节点指针。
package lru
import "sync"
type entry[K comparable, V any] struct {
key K
val V
prev, next *entry[K, V]
}
type Cache[K comparable, V any] struct {
mu sync.Mutex
capacity int
items map[K]*entry[K, V]
head *entry[K, V] // 哨兵，head.next 是最近使用的
tail *entry[K, V] // 哨兵，tail.prev 是最久未使用的
}
func New[K comparable, V any](capacity int) *Cache[K, V] {
if capacity <= 0 {
panic("lru: capacity must be positive")
}
c := &Cache[K, V]{
capacity: capacity,
items: make(map[K]*entry[K, V], capacity),
head: &entry[K, V]{},
tail: &entry[K, V]{},
}
c.head.next = c.tail
c.tail.prev = c.head
return c
}
// unlink 把节点从链表摘出来，不碰 map
func (c *Cache[K, V]) unlink(e *entry[K, V]) {
e.prev.next = e.next
e.next.prev = e.prev
}
// pushFront 把节点插到 head 后面，表示「刚刚被用过」
func (c *Cache[K, V]) pushFront(e *entry[K, V]) {
e.next = c.head.next
e.prev = c.head
c.head.next.prev = e
c.head.next = e
}
func (c *Cache[K, V]) Get(key K) (V, bool) {
c.mu.Lock()
defer c.mu.Unlock()
e, ok := c.items[key]
if !ok {
var zero V
return zero, false
}
// 命中就要挪到头部，否则整个结构退化成 FIFO
c.unlink(e)
c.pushFront(e)
return e.val, true
}
func (c *Cache[K, V]) Put(key K, val V) {
c.mu.Lock()
defer c.mu.Unlock()
// 已存在的 key 必须复用原节点，不能 new 一个新的
if e, ok := c.items[key]; ok {
e.val = val
c.unlink(e)
c.pushFront(e)
return
}
e := &entry[K, V]{key: key, val: val}
c.items[key] = e
c.pushFront(e)
// 先插入再判断，更新路径上不会算错
if len(c.items) > c.capacity {
oldest := c.tail.prev
c.unlink(oldest)
delete(c.items, oldest.key)
// 断开引用，value 是大对象时能让 GC 早点回收
oldest.prev, oldest.next = nil, nil
}
}
func (c *Cache[K, V]) Len() int {
c.mu.Lock()
defer c.mu.Unlock()
return len(c.items)
}
一个最小测试，覆盖「命中会改变淘汰顺序」这件事：
func TestEvict(t *testing.T) {
c := New[string, int](2)
c.Put("a", 1)
c.Put("b", 2)
c.Get("a") // a 变成最近使用
c.Put("c", 3) // 应该淘汰 b 而不是 a
if _, ok := c.Get("b"); ok {
t.Fatal("b should be evicted")
}
if v, _ := c.Get("a"); v != 1 {
t.Fatal("a should survive")
}
}
需要 Go 1.18+。老项目把泛型换成 interface{}，逻辑一字不改，只是取值要类型断言。
Java 其实不用手写，LinkedHashMap 构造函数第三个参数 accessOrder=true 就是维护访问顺序的双向链表：
public class LRUCache<K, V> extends LinkedHashMap<K, V> {
private final int capacity;
public LRUCache(int capacity) {
super(capacity, 0.75f, true); // true = 按访问顺序排序
this.capacity = capacity;
}
@Override
protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
return size() > capacity;
}
}
但 LinkedHashMap 不是线程安全的。套 Collections.synchronizedMap 之后每次 get 都要抢同一把锁，热点 key 上会直接打成瓶颈，见坑四。
坑一：map 里存值，淘汰时摘不掉节点
最典型的第一版错误是 m map[string]string，或者 m map[string]*Value（值类型指针，不是链表节点）。淘汰的时候你手里只有 key，没有节点，只能从头遍历链表找前驱，淘汰从 O(1) 变成 O(n)。
容量 10 万、QPS 6000 的话，每次淘汰扫 10 万个节点是几百微秒量级，而且这一下正好发生在 Put 的锁里，其他所有请求跟着排队，延迟毛刺就是这么来的。
还有一个变体：map 存节点，但 Put 更新已有 key 时 new 了一个新节点。旧节点还挂在链表上，于是同一个 key 在链表里出现两次，内存泄漏；淘汰时删的 key 又对不上，删了个寂寞。Put 必须先查 map，命中就复用节点。
坑二：Get 命中不挪位置，LRU 变 FIFO
把 Get 里的 unlink + pushFront 删掉，代码照样编译、照样跑，如果你的测试只覆盖了「不存在的 key 返回 false」，单测全绿。但淘汰顺序变成了纯先进先出。
后果是热点 key 会被后来的冷 key 按插入顺序挤掉，命中率明显下降。差距多大取决于访问分布，在 Zipf(α≈1) 的模拟里通常是十几个到二十几个百分点。建议拿一份真实访问日志，用同一个 key 序列分别喂给 FIFO 和 LRU 实现，把命中率打出来对比 —— 这比看代码可靠得多，也能顺便验证你线上容量到底该设多少。
坑三：边界条件
- capacity <= 0：别静默变成「永不缓存」，直接 panic。我在测试环境见过容量从配置读成 0 的服务，缓存全程不生效，压测数据好看得离谱，上线才炸。
- 淘汰时机：先插入再判断
len > capacity，还是先判断再插入？前者简单且正确，最多瞬时超一个；后者在「更新已有 key」的路径上很容易把容量算错，提前淘汰一个不该淘汰的。 - 容量为 1：
head.next和tail.prev是同一个节点。哨兵写法天然正确，用 if-else 判断空表/单节点的手写版本这里最容易崩。 - 淘汰后断引用：
oldest.prev, oldest.next = nil, nil不是必须，但在大 value 场景能让 GC 早点回收，也能避免调试时顺着指针翻出一堆本该释放的对象。
坑四：并发下的锁粒度
上面整个 Cache 一把 sync.Mutex。我在 8 核机器上用 b.RunParallel、读写比 9:1 压了下（key 是 16 字节字符串），单锁版本大概每秒几百万次操作，看着还行；但只要 Get 里多一点点工作（value 是结构体要拷贝、或者临界区里还做了序列化），数字立刻掉一个量级，因为所有核都在抢同一把锁。换 sync.RWMutex 没用 —— LRU 的 Get 会改链表，本质是写操作。
真正有效的是分片，每片独立加锁：
type ShardedCache[V any] struct {
shards []*Cache[string, V]
mask uint64
}
func NewSharded[V any](capacity, shards int) *ShardedCache[V] {
// shards 必须是 2 的幂，才能用位与代替取模
if shards&(shards-1) != 0 {
panic("shards must be power of two")
}
per := capacity / shards
if per == 0 {
per = 1
}
s := &ShardedCache[V]{
shards: make([]*Cache[string, V], shards),
mask: uint64(shards - 1),
}
for i := range s.shards {
s.shards[i] = New[string, V](per)
}
return s
}
func (s *ShardedCache[V]) shard(key string) *Cache[string, V] {
// FNV-1a，够快；别在这里用 md5/sha，哈希比缓存访问本身还贵
h := uint64(14695981039346656037)
for i := 0; i < len(key); i++ {
h ^= uint64(key[i])
h *= 1099511628211
}
return s.shards[h&s.mask]
}
func (s *ShardedCache[V]) Get(key string) (V, bool) { return s.shard(key).Get(key) }
func (s *ShardedCache[V]) Put(key string, val V) { s.shard(key).Put(key, val) }
代价要说清楚：总容量变成 per * shards，各分片独立淘汰。如果 key 分布倾斜（比如某几个热门商品占了 30% 流量），它们落在同一片，仍然会争锁。分片数 16、32 比较常见，再往上单核收益递减，而且每片都淘汰不满、容量碎片化更明显。
坑五：按条数淘汰，不按内存淘汰
10 万条 × 平均 20KB ≈ 2GB。但容器 limit 可能只有 1GB，而且 P99 的 value 是 200KB，不是平均值。你按条数算得好好的，OOMKilled 照样来。
要么加权重按字节淘汰：给 Put 加一个 weight 参数（通常就是 len(serialized)），维护 curWeight，淘汰循环改成 for c.curWeight > c.maxWeight。要么更省事：设一个 value 大小上限，超了直接不进缓存。
const maxValueSize = 64 << 10 // 64KB
func (s *ShardedCache[[]byte]) PutBytes(key string, v []byte) {
if len(v) > maxValueSize {
return // 大对象不进缓存，否则会把一堆小热 key 全冲掉
}
s.Put(key, v)
}
这一行判断的收益，通常比把容量从 10 万调到 20 万大得多。大对象进 LRU 会挤掉大量小热 key，这是 LRU 的经典缺陷，后面还会说到。
坑六：LRU 挡不住一次性扫描
LRU 的前提假设是「最近用过的还会再用」。爬虫、后台导出、运营全表扫这类流量不满足这个假设：它们塞进来几万个只用一次的 key，把真正的热数据全挤出去。现象是命中率周期性掉到接近 0，DB 出现规律性尖刺，而 CPU、QPS 都正常，很难查。
这不是实现 bug，是算法选型问题。三个方向：
| 方案 | 思路 | 适合 | 代价 |
|---|---|---|---|
| Segmented LRU（groupcache） | 分 probation / protected 两段，新 key 先进 probation，被二次访问才升级 | 冷热分层明显 | 实现复杂度翻倍，两段容量比例要调 |
| LRU-K / 2Q | 记录最近 K 次访问，次数不够不进主缓存 | 扫描流量多 | 维护额外历史队列，内存开销上升 |
| TinyLFU（ristretto） | Count-Min Sketch 统计频率，准入时和待淘汰者比频率 | 高频长尾、命中率优先 | 有统计误差，sketch 要定期衰减 |
经验是：单机、纯本地、几十万条以内，手写 LRU 够用。一旦命中率被扫描流量打崩，先试 value 大小限制 + 业务层按 key 前缀分流（把会扫描的维度单独放一个实例或干脆不缓存），再考虑换算法。上 TinyLFU 之前先量清楚命中率掉了多少 —— 很多时候问题不在淘汰算法，在 key 的设计，比如把不该拼的维度拼进了 cache key。
坑七：把内部结构返回出去
Go 里如果把 *entry 返回给调用方，对方随手改了 e.val 或者 e.next，链表就断了，之后所有淘汰都算错。这类 bug 要等到容量满了才暴露，而且现场看着像随机的。所以 Get 签名用 (V, bool)，返回值拷贝。
如果 V 本身是指针、map、slice，拷贝的只是引用，调用方改内容缓存里也跟着变。要么约定缓存内部对象只读，要么 Put 时深拷贝。深拷贝的代价通常是单次访问耗时的 2~3 倍，自己权衡 —— 我一般选约定只读，然后在 code review 里盯着。
什么时候别自己写
现成的轮子已经够用了。Go 有 github.com/hashicorp/golang-lru/v2（同时提供 LRU 和 2Q，泛型接口，跟上面的 New/Get/Put 几乎一样）、github.com/dgraph-io/ristretto（TinyLFU + 采样淘汰，命中率一般更好，但内部有异步 goroutine，关闭时要 Wait()）。Java 生产环境基本用 Caffeine（W-TinyLFU）。Python 是 collections.OrderedDict 配 move_to_end，或者直接用 functools.lru_cache。
自己写的理由只有两个：需要按字节权重淘汰，或者需要和已有的存储、序列化逻辑贴得很紧。其余情况用库，把精力留给命中率监控 —— 一个没有命中率指标的缓存，本质上就是个内存泄漏。