Netflix 并发 TTL Map：Lazy Cleanup、容量与 LRU

题目重点：对应 Netflix 并发 Map 汇总题的 TTL、懒清理、容量与 LRU 追问，给出单锁正确性基线，并量化过期扫描及热点锁成本。

题意与假设

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

训练版 API 是 get/put(key,value,ttl)，ttl 从 put 时开始、使用 monotonic 时钟；到 `now>=expires_at` 即过期。get 不刷新 TTL，但刷新 LRU。
容量按 entry 数计；put 先删除过期项，若仍满，淘汰最近最少访问且未过期的一项。ttl=0 表示立即不存。
get 缺失/过期返回 None，因此训练 API 不区分已存 value=None 与未命中；真实接口可用 sentinel 或 `(found,value)`。

解题路线

用一把锁包住 OrderedDict 的读取、过期判定、移动到尾部和 put 淘汰。这是线性化的正确性基线，避免两个线程同时占最后一个容量或删除正在刷新的 key。
get 只惰性检查被访问的 key；put 为腾容量扫描当前 entries 并删所有过期项，然后更新目标 key，超容量时 pop 最旧项。若不清过期就直接 LRU，可能错误驱逐有效 entry。
高容量/高写入场景当前 put 扫描 O(C)；可用 expiry min-heap 加版本号惰性失效降低过期清理均摊成本，或按时间轮近似清理。并发吞吐可按 key 分片，但全局精确 LRU 与全局容量要额外协调。

手动走一遍

训练例容量 2，put A TTL=1、B TTL=100，时钟走到 2 时 put C：先删过期 A，再插 C，因此不应淘汰仍有效的 B。get(B) 刷新其 LRU 位置，但不会把原过期时间从 100 延后。

Python 参考实现

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

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

from collections import OrderedDict
from threading import Lock
from time import monotonic


class ConcurrentTTLMap:
    """Bounded LRU map with per-entry absolute expiry and one mutex."""

    def __init__(self, capacity, clock=monotonic):
        if capacity <= 0:
            raise ValueError("capacity must be positive")
        self._capacity = capacity
        self._clock = clock
        self._entries = OrderedDict()
        self._lock = Lock()

    def get(self, key):
        with self._lock:
            entry = self._entries.get(key)
            if entry is None:
                return None
            value, expires_at = entry
            if self._clock() >= expires_at:
                del self._entries[key]
                return None
            self._entries.move_to_end(key)
            return value

    def put(self, key, value, ttl):
        if ttl < 0:
            raise ValueError("negative ttl")
        with self._lock:
            now = self._clock()
            self._entries.pop(key, None)
            if ttl == 0:
                return
            for old_key, (_, expires_at) in list(self._entries.items()):
                if now >= expires_at:
                    del self._entries[old_key]
            self._entries[key] = (value, now + ttl)
            if len(self._entries) > self._capacity:
                self._entries.popitem(last=False)

复杂度

get 预期 O(1)，put 当前实现扫描最多 C 条并构造列表，O(C) 时间/空间；entries 空间 O(C)。单锁串行所有操作。生产优化 expiry heap 的每次插入/清理约 O(log C)，但要处理旧版本 heap 节点与全局 LRU 同步。

易错点与追问

容易失分的边界

按墙上时钟过期，系统时间回拨使 entry 复活。
get 先刷新 LRU 再检查 TTL，让过期项占热点位置。
容量满时先淘汰有效最旧项，之后才发现有过期项。
两个线程分别检查容量后同时插入，最终超限。
声称 lazy cleanup 是精确 O(1) put，却全表扫描过期项。

继续深挖

若容量按字节而非条数，如何原子更新权重和淘汰？
get 是否应延长 TTL，语义改变在哪里？
全局 LRU 在分布式分片上如何近似？
后台清理与前台 get/put 如何避免双重删除和锁争用？