如何实现一个带 TTL 的简单 dict 缓存(时间过期)

Python自带dict不支持TTL因其纯内存映射,无时间戳、过期检查或自动清理机制;手动维护易致内存泄漏,Timer方案开销大且难管理;推荐懒过期封装方案,读时检查并删除过期项。

为什么 Python 自带的 dict 不能直接支持 TTL

因为 dict 是纯内存映射结构,没有内置时间戳、过期检查或后台清理机制。你写入一个键值对后,它就一直存在,哪怕逻辑上该数据已“过期”。手动维护时间戳虽可行,但每次读取都得判断 time.time() > expire_at,且不会自动删除——久而久之会内存泄漏。

threading.Timer 做延迟删除?不推荐

为每个 key 启一个 Timer 看似直观,但实际问题很多:

  • 大量短生命周期 key 会创建海量线程对象,开销大、难管理
  • Timer 一旦启动无法取消(除非用第三方库如 threading.Timer.cancel() 配合状态标记,但易出竞态)
  • 进程退出时未触发的 Timer 不会自动释放,可能卡住程序
  • 无法统一控制最大缓存数或 LRU 淘汰

所以,除非 key 极少且生命周期极长,否则别走这条路。

推荐方案:封装 dict + 时间戳 + 懒过期(lazy expiration)

这是最轻量、可控、无依赖的做法,适合大多数内部工具或脚本场景。核心思路是:写入时记录过期时间,读取时检查是否过期,过期则删掉并返回 None 或抛异常。

示例实现:

import time

class TTLCache:
    def __init__(self):
        self._cache = {}
        self._times = {}

    def set(self, key, value, ttl_seconds):
        self._cache[key] = value
        self._times[key] = time.time() + ttl_seconds

    def get(self, key, default=None):
        expire_at = self._times.get(key)
        if expire_at is None:
            return default
        if time.time() > expire_at:
            self._cache.pop(key, None)
            self._times.pop(key, None)
            return default
        return self._cache[key]

    def delete(self, key):
        self._cache.pop(key, None)
        self._times.pop(key, None)

关键点:

  • 不依赖外部库,纯标准库,Python 3.6+
  • 读操作触发过期检查(懒过期),避免定时扫描开销
  • 写操作只存两个 dict,无锁(单线程安全;多线程需加 threading.Lock
  • ttl_seconds 为浮点数,支持亚秒级精度(如 0.1

需要线程安全?加一把 threading.Lock 就够了

如果多个线程共用同一个 TTLCache 实例,必须保护共享状态。只需在关键方法加锁,不需要全量同步:

import threading

class ThreadSafeTTLCache:
    def __init__(self):
        self._cache = {}
        self._times = {}
        self._lock = threading.Lock()

    def set(self, key, value, ttl_seconds):
        with self._lock:
            self._cache[key] = value
            self._times[key] = time.time() + ttl_seconds

    def get(self, key, default=None):
        with self._lock:
            expire_at = self._times.get(key)
            if expire_at is None:
                return default
            if time.time() > expire_at:
                self._cache.pop(key, None)
                self._times.pop(key, None)
                return default
            return self._cache[key

]

注意:getset 都要锁,但不必锁整个生命周期——比如不要在 get 里长时间持有锁去处理业务逻辑。

真正容易被忽略的是:懒过期机制下,过期 key 可能长期滞留在内存中,直到下次 get 或显式 delete。如果写多读少(比如埋点日志类缓存),得配合定期调用 cleanup() 扫描清理,或者换用 heapq 维护最小堆做主动过期。