本文目录导读：
Python中实现缓存淘汰策略（Cache Eviction Policies）通常有以下几种常见算法，下面是每种策略的详细实现：
LRU (最近最少使用) 策略
使用 OrderedDict 实现
from collections import OrderedDict
import time
class LRUCache:
def __init__(self, capacity: int):
self.cache = OrderedDict()
self.capacity = capacity
def get(self, key):
if key not in self.cache:
return -1
# 移动到末尾（最近使用）
self.cache.move_to_end(key)
return self.cache[key]
def put(self, key, value):
if key in self.cache:
# 更新并移动到末尾
self.cache.move_to_end(key)
elif len(self.cache) >= self.capacity:
# 淘汰最久未使用的（第一个元素）
self.cache.popitem(last=False)
self.cache[key] = value
使用双链表 + 哈希表实现（更高效）
class DLinkedNode:
def __init__(self, key=0, value=0):
self.key = key
self.value = value
self.prev = None
self.next = None
class LRUCacheOptimized:
def __init__(self, capacity: int):
self.capacity = capacity
self.cache = {}
# 虚拟头尾节点
self.head = DLinkedNode()
self.tail = DLinkedNode()
self.head.next = self.tail
self.tail.prev = self.head
def _add_to_head(self, node):
node.prev = self.head
node.next = self.head.next
self.head.next.prev = node
self.head.next = node
def _remove_node(self, node):
node.prev.next = node.next
node.next.prev = node.prev
def _move_to_head(self, node):
self._remove_node(node)
self._add_to_head(node)
def _pop_tail(self):
node = self.tail.prev
self._remove_node(node)
return node
def get(self, key):
if key not in self.cache:
return -1
node = self.cache[key]
self._move_to_head(node)
return node.value
def put(self, key, value):
if key in self.cache:
node = self.cache[key]
node.value = value
self._move_to_head(node)
else:
if len(self.cache) >= self.capacity:
# 淘汰尾部节点
tail = self._pop_tail()
del self.cache[tail.key]
new_node = DLinkedNode(key, value)
self.cache[key] = new_node
self._add_to_head(new_node)
LFU (最不经常使用) 策略
class LFUCache:
def __init__(self, capacity: int):
self.capacity = capacity
self.key_to_node = {} # key -> node
self.freq_to_nodes = {} # freq -> OrderedDict
self.min_freq = 0
def _update_freq(self, key):
node = self.key_to_node[key]
old_freq = node.freq
freq_dict = self.freq_to_nodes[old_freq]
# 从旧频率字典移除
del freq_dict[key]
if not freq_dict:
del self.freq_to_nodes[old_freq]
if old_freq == self.min_freq:
self.min_freq += 1
# 更新到新频率
node.freq += 1
new_freq = node.freq
if new_freq not in self.freq_to_nodes:
self.freq_to_nodes[new_freq] = OrderedDict()
self.freq_to_nodes[new_freq][key] = node
def get(self, key):
if key not in self.key_to_node:
return -1
self._update_freq(key)
return self.key_to_node[key].value
def put(self, key, value):
if self.capacity <= 0:
return
if key in self.key_to_node:
node = self.key_to_node[key]
node.value = value
self._update_freq(key)
else:
if len(self.key_to_node) >= self.capacity:
# 淘汰最不经常使用的
min_freq_dict = self.freq_to_nodes[self.min_freq]
k, _ = min_freq_dict.popitem(last=False)
del self.key_to_node[k]
if not min_freq_dict:
del self.freq_to_nodes[self.min_freq]
# 添加新节点
new_node = Node(key, value)
self.key_to_node[key] = new_node
self.min_freq = 1
if 1 not in self.freq_to_nodes:
self.freq_to_nodes[1] = OrderedDict()
self.freq_to_nodes[1][key] = new_node
class Node:
def __init__(self, key, value):
self.key = key
self.value = value
self.freq = 1
FIFO (先进先出) 策略
from collections import deque
class FIFOCache:
def __init__(self, capacity: int):
self.capacity = capacity
self.cache = {}
self.queue = deque()
def get(self, key):
return self.cache.get(key, -1)
def put(self, key, value):
if key not in self.cache:
if len(self.cache) >= self.capacity:
# 淘汰最早插入的
oldest_key = self.queue.popleft()
del self.cache[oldest_key]
self.queue.append(key)
self.cache[key] = value
TTL (生存时间) 策略
import time
import threading
class TTLCache:
def __init__(self, capacity: int, default_ttl: int = 300):
self.capacity = capacity
self.default_ttl = default_ttl
self.cache = {}
self.lock = threading.Lock()
# 启动清理线程
self._cleanup_thread = threading.Thread(target=self._periodic_cleanup, daemon=True)
self._cleanup_thread.start()
def _is_expired(self, timestamp, ttl):
return time.time() - timestamp > ttl
def _periodic_cleanup(self):
"""定期清理过期缓存"""
while True:
time.sleep(60) # 每分钟清理一次
with self.lock:
current_time = time.time()
expired_keys = [
k for k, v in self.cache.items()
if self._is_expired(v['timestamp'], v['ttl'])
]
for key in expired_keys:
del self.cache[key]
def get(self, key):
with self.lock:
if key not in self.cache:
return None
entry = self.cache[key]
if self._is_expired(entry['timestamp'], entry['ttl']):
del self.cache[key]
return None
return entry['value']
def put(self, key, value, ttl=None):
if ttl is None:
ttl = self.default_ttl
with self.lock:
if key not in self.cache and len(self.cache) >= self.capacity:
# 淘汰过期条目，或最旧的
current_time = time.time()
# 先尝试淘汰过期条目
expired_keys = [
k for k, v in self.cache.items()
if self._is_expired(v['timestamp'], v['ttl'])
]
if expired_keys:
del self.cache[expired_keys[0]]
else:
# 淘汰最旧的
oldest_key = min(self.cache.keys(), key=lambda k: self.cache[k]['timestamp'])
del self.cache[oldest_key]
self.cache[key] = {
'value': value,
'timestamp': time.time(),
'ttl': ttl
}
使用装饰器实现缓存（简单版）
from functools import wraps
import time
def lru_cache(maxsize=128):
def decorator(func):
cache = OrderedDict()
@wraps(func)
def wrapper(*args, **kwargs):
key = (args, tuple(kwargs.items()))
if key in cache:
cache.move_to_end(key)
return cache[key]
result = func(*args, **kwargs)
if len(cache) >= maxsize:
cache.popitem(last=False)
cache[key] = result
return result
return wrapper
return decorator
# 使用示例
@lru_cache(maxsize=32)
def expensive_function(n):
# 模拟耗时计算
time.sleep(1)
return n * n
使用建议
- 简单场景：使用
functools.lru_cache装饰器 - 需要精确控制：自定义 LRU 实现
- 数据访问模式：
- 顺序访问：FIFO
- 热点访问：LRU
- 频率优先：LFU
- 需要过期时间：TTL 策略
- 缓存穿透防护：使用布隆过滤器配合
选择哪个策略取决于你的具体应用场景和数据访问模式。
