如何不使用第三方库实现一个简单的线程安全 LRU 缓存

10次阅读

可用python内置threading模块配合OrderedDict实现线程安全LRU缓存,核心是用Lock保护OrderedDict的move_to_end和popitem操作,确保访问更新与超容淘汰的原子性。

如何不使用第三方库实现一个简单的线程安全 LRU 缓存

可以用 Python 内置的 threading 模块配合 OrderedDict(或手动维护链表+字典)实现一个轻量、线程安全的 LRU 缓存,无需第三方库。

核心思路:锁 + 有序结构

LRU 的关键在于「访问时更新顺序」和「容量超限时淘汰最久未用项」。Python 标准库collections.OrderedDict 天然支持 move_to_end(key)popitem(last=False),正好对应「标记最近使用」和「淘汰最老项」。线程安全只需对所有读写操作加一把互斥锁即可。

基础实现(基于 OrderedDict)

以下是一个完整、可直接运行的版本:

from collections import OrderedDict import threading 

class ThreadSafeLRUCache: def init(self, capacity: int): self.capacity = capacity self._cache = OrderedDict() self._lock = threading.Lock()

def get(self, key):     with self._lock:         if key not in self._cache:             return None         # 移到末尾,表示最近被访问         self._cache.move_to_end(key)         return self._cache[key]  def put(self, key, value):     with self._lock:         if key in self._cache:             self._cache.move_to_end(key)         self._cache[key] = value         # 超容时删最老的(开头)         if len(self._cache) > self.capacity:             self._cache.popitem(last=False)  def size(self):     with self._lock:         return len(self._cache)  def clear(self):     with self._lock:         self._cache.clear()

关键细节说明

  • 锁粒度合理:每个方法内部独占锁,避免并发读写导致 OrderedDict 状态不一致(如 get 中判断存在后、move_to_end 前被另一个线程删除)
  • 不依赖 get 的返回值做逻辑分支:例如不要写 if cache.get(k) is None: cache.put(k, v),这会引发两次加锁,且中间可能被其他线程修改。如需原子性检查并设置,应新增 setdefault 类方法
  • capacity ≤ 0 时行为明确:可在 __init__ 中校验,或让 put 直接忽略插入(当前实现会允许存但立即淘汰,等效于只读缓存)
  • 避免在锁内做耗时操作:比如不建议在 get 中触发网络请求或复杂计算;缓存本身只负责快速存取

进阶:无 OrderedDict 的纯字典 + 双向链表(兼容旧 Python)

如果运行环境是 Python dict 无序),或想彻底掌控结构,可用字典 + 手写双向链表节点。但实现更重:

  • 每个键对应一个链表节点,含前后指针和值
  • 字典映射 key → node,便于 O(1) 定位
  • 维护 head(最新)和 tail(最老)两个哨兵节点
  • get:查字典 → 摘下节点 → 插入 head 后 → 返回值
  • put:已存在则更新值并移到 head;不存在则新建节点插 head 后,若超容则删 tail 前一个节点并从字典移除

该方案性能略优(免去 OrderedDict 的部分封装开销),但代码量翻倍,调试成本高。一般场景推荐用 OrderedDict 版本。

text=ZqhQzanResources