Amazon Cache：TTL + LRU，主动过期怎样避免旧计时器误删？

题意与假设

这道练习题保留了面经中值得训练的核心约束，并将需要澄清的条件单独列出。现场拿到变体时，先和面试官确认这些条件。

• 训练单线程，正容量，时间非降序；get成功刷新LRU但不续TTL，put更新续TTL并变MRU。
• deadline≤now视为过期，TTL=0立即删除同key，ttl=None永不过期；get返回(found,value)区分存None和miss。
• advance(now)供定时驱动调用；题目并未要求此API与时间单位，生产中应有单调时钟与锁。

解题路线

1. 状态分工：OrderedDict从LRU到MRU维护活跃entry，过期堆存(deadline,version,key)。每次put生成全局唯一version，get/put前advance先清到期再做容量判断。
2. 堆记录不立即移除旧版本：更新或淘汰后到期记录仍可在heap，但只有当前key的version等于记录才允许删entry。相同deadline也靠唯一version稳定比较，key无需互相可排序。
3. active expiry定时器看heap最早deadline安排下一次advance，过期成本与到期记录数相关，避免每轮全表扫描；回调仍应核版本，取消timer可能与触发竞争。
4. 容量淘汰在清过期后进行，写入新值MRU，超限删除LRU。只在get时清所取key会令其他过期key占容量，错误淘汰未过期值。
5. 大量刷新同key会积累无效heap记录，代码在记录数>max(64,2·活跃数)时按现值重建。峰值O(capacity+64)，重建有O(capacity)延迟，不能声称每次put严格O(log capacity)。
6. 生产并发需在同一锁/事务保护expiry、LRU和容量，定时器与请求遵同一状态规则。多进程全局cache不能只靠线程锁；读取TTL通常不意味着续期。

手动走一遍

训练put(a,1,t=0,ttl=5)，t=2更新a=2,ttl=10；t=5旧记录到期不得删除新a，t=12才过期。容量2时过期b应先清，不应该为了新c淘汰仍活跃的a。

Python 参考实现

下面的代码按上文假设实现，适合本地运行并追加自己的测试用例。

代码区可横向滚动；使用键盘时，先聚焦代码区再按方向键查看。

from collections import OrderedDict
from heapq import heappop,heappush,heapify


class TTLLRU:
    """Single-threaded monotonic time; get returns (found,value). get does not renew TTL."""
    def __init__(self,capacity):
        if capacity <= 0:raise ValueError("capacity must be positive")
        self.capacity=capacity
        self.entries=OrderedDict()
        self.expirations=[]
        self.version=0
        self.last_time=None

    def advance(self,now):
        if self.last_time is not None and now < self.last_time:
            raise ValueError("time must be nondecreasing")
        self.last_time=now
        while self.expirations and self.expirations[0][0] <= now:
            deadline,version,key=heappop(self.expirations)
            current=self.entries.get(key)
            if current is not None and current[2]==version:
                del self.entries[key]
        self._compact()

    def _compact(self):
        if len(self.expirations) > max(64,2*len(self.entries)):
            self.expirations=[(deadline,version,key)
                              for key,(value,deadline,version) in self.entries.items()
                              if deadline is not None]
            heapify(self.expirations)

    def put(self,key,value,now,ttl=None):
        if ttl is not None and ttl < 0:raise ValueError("negative TTL")
        self.advance(now)
        self.entries.pop(key,None)
        if ttl==0:
            return
        self.version+=1
        deadline=None if ttl is None else now+ttl
        self.entries[key]=(value,deadline,self.version)
        if deadline is not None:
            heappush(self.expirations,(deadline,self.version,key))
        while len(self.entries)>self.capacity:
            self.entries.popitem(last=False)
        self._compact()

    def get(self,key,now):
        self.advance(now)
        if key not in self.entries:return False,None
        self.entries.move_to_end(key)
        return True,self.entries[key][0]

复杂度

普通命中LRU操作O(1)，另加k条到期堆记录的O(k log H)清理；put堆插入O(log H)，偶尔O(capacity)重建。H被压缩阈值控制；定时精度决定内存过期清理延迟，不改变deadline读语义。

易错点与追问

容易失分的边界

• 旧到期事件按key误删更新值。
• LRU刷新访问时顺便隐式续TTL。
• 过期占容量导致误淘汰。
• heap永不压缩导致刷新内存增长。

继续深挖

• TTL刷新on-read需怎样生成新版本？
• 多个shard怎样分配容量？
• 定时器阻塞时读请求怎样保持过期正确？