如何实现一个带过期时间的简单内存缓存(不依赖 cachetools)

核心思路是用dict+threading.Timer实现键过期:写入时存值并启动定时删除,用RLock保护并发访问,覆盖key前需cancel旧timer,get时需检查timer.is_alive(),ttl统一转为秒数,严格配对cancel与pop防泄漏。

dict + threading.Timer 实现基础过期逻辑

核心思路是:每次写入时启动一个定时器,在过期时间到达后自动删除键。这不是线程安全的纯字典操作,但比轮询或后台清理更轻量。

  • 每次调用 set(key, value, ttl) 时,先存入 self._cache[key] = value,再创建并启动一个 threading.Timer(ttl, self._delete_key, args=[key])
  • 需用 self._timers 字典保存当前活跃的定时器(key → Timer 实例),避免重复设置导致旧定时器未取消而误删
  • 写入同 key 时必须先 .cancel() 旧定时器,否则可能残留已过期但尚未触发的定时任务
  • 注意 Timer 启动后无法修改,只能 cancel + 新建;且 cancel 在 timer 已触发后无副作用,所以判断是否存活要用 is_alive() 配合状态管理

处理并发读写时的竞态问题

多个线程同时 getset 同一个 key 可能导致缓存不一致或定时器泄漏。

  • threading.RLock 包裹所有对 self._cacheself._timers 的访问,包括 getset_delete_key
  • get 中不能只检查 key 是否存在,还要确认对应定时器是否仍在运行(self._timers.get(key, None) and timer.is_alive()),否则可能读到已被 cancel 但尚未触发删除的脏值
  • 避免在 get 里重置 TTL —— 这属于 LRU 行为,和“固定过期时间”语义冲突;如需刷新过期时间,应显式调用 set

ttl 参数支持秒级浮点数与 datetime.timedelta

用户传入 timedelta(seconds=30)30.5 都应被接受,内部统一转为 float 秒数。

  • set 方法开头做类型归一化:if isinstance(ttl, timedelta): ttl = ttl.total_seconds()
  • 注意 timedelta 可能为负(表示立即过期),此时直接跳过设值和定时器启动
  • 不建议支持字符串如 "30s" —— 增加解析开销且易出错,交由上层转换更清晰

内存泄漏风险:未 cancel 的 Timer 对象

会持引用

Python 的 threading.Timer 在触发前会强引用其回调函数和参数。若忘记 cancel,即使缓存 key 被删,Timer 仍存活并阻止对象回收。

  • 务必在 set 覆盖旧 key 前执行 old_timer.cancel(),并在 _delete_key 执行后从 self._timers 中 pop 掉该 key
  • 可加一个简单的调试钩子:在 __del__ 或关闭时遍历 self._timers.values(),打印仍存活的 timer 数量(仅开发用)
  • 真正长期运行的服务建议搭配弱引用或定期扫描 _timers 中已失效 timer,但对简单缓存来说,严格配对 cancel + pop 已足够
实际使用中,最常被忽略的是定时器 cancel 的时机和 is_alive() 的判断位置 —— 它们必须在同一个锁保护下完成,否则仍存在微小窗口期导致 double-delete 或漏 cancel。