cswork / 系统设计

第 18 章 / 共 57 章 · 约 5 分钟

LLD 案例 2:带 TTL 的 LRU 缓存

本页目录
  1. 题目与澄清
  2. 基础结构:哈希表 + 双向链表
  3. 加入 TTL:两种过期方式
  4. 淘汰顺序
  5. 代码草稿
  6. 并发:一个常见陷阱
  7. 扩展讨论
  8. 章节测验

题目与澄清

基础版是 LeetCode 146:容量固定,get / put 都要 O(1),满了淘汰最久未使用的条目。面试中常见的加码是:每个条目带过期时间(TTL),过期后对外不可见。

先问清楚:

  • TTL 是全局统一,还是每次 put 时指定?(按每条指定来设计,统一 TTL 只是特例)
  • 过期条目算不算容量?(要算,内存是实打实占着的)
  • put 一个已存在的 key,是否刷新 TTL?(通常是)
  • 需要线程安全吗?(先单线程写对,再讨论并发)
  • size() 需要精确吗?(这决定了要不要主动清理)

基础结构:哈希表 + 双向链表

HashMap: key -> Node

head <-> [k3] <-> [k1] <-> [k7] <-> tail
        最近使用            最久未使用
  • 哈希表负责 O(1) 定位节点。
  • 双向链表负责 O(1) 调整顺序:访问时把节点摘下来挪到头部;淘汰时删尾部。
  • 头尾各放一个哨兵节点,插入删除不用判空,代码能短一半,也不容易写错。

为什么必须是双向链表?删除任意节点需要拿到它的前驱,单链表要 O(n) 才能找到前驱。

加入 TTL:两种过期方式

惰性过期:get 时检查 expireAt <= now,过期就删掉并返回未命中。实现简单,没有额外开销。

缺点:一个过期但再也没人访问的条目,会一直占着容量和内存,直到被 LRU 挤到尾部淘汰。对多数场景这可以接受,因为 LRU 本身就会把冷数据挤出去。

主动清理:额外维护一个按 expireAt 排序的最小堆。put 时先弹出所有已过期的条目,再判断是否需要按 LRU 淘汰。

minHeap: (100, k3)  (130, k1)  (200, k7)
          ^ 最早过期

堆不支持高效地删除任意元素,所以用惰性删除:堆里记录 (expireAt, key),弹出时去哈希表核对——如果这个 key 已经不在,或者它当前的 expireAt 和堆里记的不一致(说明被重新 put 过),就直接丢弃这个堆元素。

惰性删除有一个泄漏点:同一个 key 被反复 put 且 TTL 很长时,每次都往堆里压一条新记录,旧记录的过期时间都还没到,弹不出来,堆会无限增长,远远超过缓存容量。补救办法是:堆的大小超过哈希表大小的 2 倍时,用哈希表里的当前条目重建堆。重建是 O(n),但至少积累 n 条过时记录才触发一次,摊还到每次 put 上只多 O(1)。

操作只做惰性过期惰性 + 最小堆
getO(1)O(1)
putO(1)O(log n) 摊还
过期条目占用容量可能长期占用put 时及时回收
实现复杂度低中

Redis 的思路介于两者之间:惰性删除,再加上定期随机抽样一批带过期时间的 key 做清理,用概率换开销。面试中提一句即可。

淘汰顺序

put 导致超容量时:

  1. 先清掉已经过期的条目(如果有堆);
  2. 仍然超容量,再从链表尾部淘汰最久未使用的条目。

先清过期的,是因为它们反正已经不可见,没有理由为了保留它们而牺牲一个仍然有效的条目。

代码草稿

面试官一般会要求手写链表,但可以先给出一个借助标准库的版本,快速确认你理解了语义:

import time
from collections import OrderedDict

class TTLCache:
    def __init__(self, capacity, clock=time.monotonic):
        self.cap, self.clock = capacity, clock
        self.data = OrderedDict()            # key -> (value, expire_at)

    def get(self, key):
        item = self.data.get(key)
        if item is None:
            return None
        value, exp = item
        if exp <= self.clock():              # 惰性过期
            del self.data[key]
            return None
        self.data.move_to_end(key)           # 标记为最近使用
        return value

    def put(self, key, value, ttl):
        self.data[key] = (value, self.clock() + ttl)
        self.data.move_to_end(key)
        if len(self.data) > self.cap:
            self.data.popitem(last=False)    # 淘汰最久未使用

几个值得主动说出的细节:

  • 用单调时钟(monotonic)。系统时间可能被 NTP 回拨或跳变,用墙上时钟会让过期判断出错。
  • 时钟可注入:测试时传一个手动推进的假时钟,验证过期不用真的 sleep。
  • 这个版本满了直接淘汰尾部,可能淘汰掉一个有效条目,却留着别处已经过期的条目——这正是引入最小堆的动机。

手写版本就是把 OrderedDict 换成“哈希表 + 带哨兵的双向链表”,move_to_end 对应“摘下节点”和“插到头部”两个辅助函数。

并发:一个常见陷阱

很多人的第一反应是读写锁:get 拿读锁,put 拿写锁。这是错的——LRU 的 get 也要移动链表节点,本质上是写操作。多个线程拿着读锁同时改链表指针,链表会被改坏。

可行方案:

  • 一把互斥锁包住 get / put。最简单、正确,中低并发下完全够用。
  • 分段:按 key 的哈希分成 N 个独立的小 LRU,每段一把锁。代价是 LRU 顺序只在段内成立,全局是近似的。
  • 近似 LRU:读操作只把访问记录写进一个缓冲区,由后台批量重放到链表上;或者像 Redis 那样随机采样淘汰。生产级缓存库常用这类思路换吞吐。

在面试中说出“读写锁不适用,因为 get 会修改访问顺序”,通常就是并发部分的关键得分点。

扩展讨论

  • 淘汰回调:条目被淘汰或过期时通知调用方,比如关闭连接、释放资源。
  • 按大小而不是条数限制容量:每个条目带 weight,一直淘汰到总权重低于上限。
  • 防缓存击穿:同一个 key 未命中时只让一个线程去加载,其他线程等待同一个结果。

章节测验

6 道选择题
  1. LRU 为什么要用双向链表而不是单链表?
  2. 为什么读写锁不适合 LRU 缓存?
  3. 某个 key 被重新 put 后,按本章的做法,最小堆里旧的过期记录怎么处理?
  4. 计算过期时为什么用单调时钟?
  5. put 导致超容量时,应该先淘汰什么?
  6. 只做惰性过期的主要缺点是?

章节讨论

正在加载讨论…