不能直接用 threading.Timer 逐个键设定时器,因为会快速耗尽线程资源、导致定时器取消困难、主线程退出时后台定时器静默失效,且多 Timer 并发 del 字典可能引发 RuntimeError;正确做法是单线程+最小堆(heapq)集中管理过期时间,插入和检查均为 O(log n)。

为什么不能直接用 threading.Timer 逐个键设定时器
每个过期键都起一个 threading.Timer 看似直观,但实际会快速耗尽线程资源——尤其当字典频繁写入、键数达千级时,Timer 对象堆积、取消困难,还可能因主线程退出导致后台定时器静默失效。更严重的是,多个 Timer 并发调用 del 同一字典可能触发 RuntimeError: dictionary changed size during iteration。
正确做法是:只启一个后台线程,统一扫描 + 清理,所有键的过期时间集中管理。
用 heapq 维护最小堆实现高效过期调度
把每个键的过期时间戳(time.time() + ttl)和键名组成元组,推入最小堆。每次循环只需检查堆顶是否已过期,避免遍历全部键,时间复杂度从 O(n) 降到 O(log n)。
- 插入时:用
heapq.heappush(self._heap, (expire_at, key)) - 清理时:循环
while self._heap and self._heap[0][0] 弹出并删除对应键 - 注意:堆中可能残留已删除键的旧记录(“脏项”),需在弹出后用
key in self._data二次校验
如何安全支持并发读写(dict 非线程安全)
Python 原生 dict 不是线程安全的,后台清理线程和用户线程同时操作会引发 RuntimeError 或数据丢失。必须加锁,但粒度要细:
立即学习“Python免费学习笔记(深入)”;
- 读操作(
__getitem__,get)只读共享数据,用self._lock.acquire(blocking=False)尝试获取锁,失败则直接返回(避免阻塞读) - 写操作(
__setitem__,__delitem__)必须阻塞获取锁:self._lock.acquire() - 清理线程中所有对
self._data和self._heap的访问,也必须持有同一把锁
别用 threading.RLock——它允许同一线程重复获取,反而掩盖了锁使用不一致的问题。
初始化时传入 cleanup_interval 的实际影响
这个参数控制后台线程的轮询间隔(单位秒),不是“精确过期时间”。设太小(如 0.01)会导致 CPU 空转;设太大(如 30)则键可能最多延迟 cleanup_interval 秒才被清理。生产环境建议 1–5 秒之间权衡。
示例初始化:
cache = ExpiringDict(ttl=60, cleanup_interval=2)
注意:如果业务对过期精度要求极高(比如秒级鉴权 Token),不应依赖后台清理,而应在每次 get() 时检查单个键的过期时间——但这会增加每次读的开销。
真正难处理的是清理线程与 del 操作的竞态:用户刚删掉一个键,后台线程恰好扫到它旧的过期记录。必须靠“懒删除 + 脏项过滤”来兜底,这点容易被忽略。


















