Python脚本缓存淘汰策略如何实现

本文目录导读：

1. LRU (最近最少使用) 策略
2. LFU (最不经常使用) 策略
3. FIFO (先进先出) 策略
4. TTL (生存时间) 策略
5. 使用装饰器实现缓存（简单版）
6. 使用建议

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

使用建议

1. 简单场景：使用 functools.lru_cache 装饰器
2. 需要精确控制：自定义 LRU 实现
3. 数据访问模式：
   • 顺序访问：FIFO
   • 热点访问：LRU
   • 频率优先：LFU
4. 需要过期时间：TTL 策略
5. 缓存穿透防护：使用布隆过滤器配合

选择哪个策略取决于你的具体应用场景和数据访问模式。