核心思路是用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()配合状态管理
处理并发读写时的竞态问题
多个线程同时 get 或 set 同一个 key 可能导致缓存不一致或定时器泄漏。
- 用
threading.RLock包裹所有对self._cache和self._timers的访问,包括get、set、_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 已足够
is_alive() 的判断位置 —— 它们必须在同一个锁保护下完成,否则仍存在微小窗口期导致 double-delete 或漏 cancel。










