如何实现一个带锁的线程安全 LRU 缓存(简单版)

9次阅读

用互斥锁保护核心操作即可实现线程安全的LRU缓存,关键在于保证get/put原子性;借助OrderedDict天然顺序特性,通过move_to_end和popitem(last=False)实现热度提升与尾部淘汰,配合threading.Lock确保并发安全。

如何实现一个带锁的线程安全 LRU 缓存(简单版)

用互斥锁保护核心操作,就能实现一个线程安全的简单 LRU 缓存。关键不是追求极致性能,而是保证 get / put 的原子性、避免数据竞争。

用双向链表 + 哈希表模拟 LRU 行为

LRU 的本质是“最近访问的放前面,容量满时淘汰尾部”。不需要自己写链表,用 pythoncollections.OrderedDict 最直接:它天然保持插入/访问顺序,且 move_to_end(key)popitem(last=False) 分别对应“提升热度”和“淘汰最久未用”。

  • 初始化时指定最大容量 capacity
  • get(key):命中则 move_to_end(key) 并返回值;未命中返回 -1
  • put(key, value):已存在则更新值并 move_to_end;不存在则插入,超容时先 popitem(last=False)

用 threading.Lock 串行化所有操作

OrderedDict 本身不是线程安全的——并发调用 move_to_endpopitem 可能导致内部状态错乱。只需在每个公有方法入口加锁,出口释放:

  • 定义一个实例属性 self._lock = threading.Lock()
  • getput 开头都写 with self._lock:
  • 锁粒度是整个方法,不拆细(简单版不考虑读写分离或分段锁)

注意边界与异常情况

容量可能为 0 或负数,key 可能为 None 或不可哈希类型,这些虽非核心逻辑,但影响健壮性:

  • 构造时若 capacity ,可设为空字典或直接抛 ValueError
  • 对 key 做基本检查:比如 if key is None: 直接 return / raise
  • put 时 value 为 None 是允许的(缓存空结果也常见),无需拦截

完整可运行示例(Python)

以下是一个去掉注释后不到 30 行的可用版本:

import threading from collections import OrderedDict  class LRUCache:     def __init__(self, capacity: int):         self.capacity = capacity         self.cache = OrderedDict()         self._lock = threading.Lock()      def get(self, key: int) -> int:         with self._lock:             if key not in self.cache:                 return -1             self.cache.move_to_end(key)             return self.cache[key]      def put(self, key: int, value: int) -> None:         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)

text=ZqhQzanResources