按使用频率建立热数据缓存

本文目录导读：

1. 核心设计方案
2. 优化建议

我来帮你设计一个基于使用频率的热数据缓存系统。

核心设计方案

基础数据结构

import heapq
from collections import defaultdict
import time
from typing import Any, Optional
class FrequencyNode:
    """频率节点"""
    def __init__(self, key: Any, value: Any, freq: int = 1):
        self.key = key
        self.value = value
        self.freq = freq
        self.last_access = time.time()
class LFUCache:
    """基于使用频率的缓存（LFU）"""
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.cache = {}  # key -> FrequencyNode
        self.freq_map = defaultdict(dict)  # freq -> {key: node}
        self.min_freq = 1
        self.access_count = 0  # 记录访问次数
    def get(self, key: Any) -> Optional[Any]:
        """获取缓存，更新频率"""
        if key not in self.cache:
            return None
        node = self.cache[key]
        self._update_frequency(node)
        node.last_access = time.time()
        return node.value
    def put(self, key: Any, value: Any):
        """设置缓存"""
        if self.capacity <= 0:
            return
        # 如果key已存在，更新值并增加频率
        if key in self.cache:
            node = self.cache[key]
            node.value = value
            self._update_frequency(node)
            return
        # 检查容量
        if len(self.cache) >= self.capacity:
            self._evict()
        # 添加新节点
        node = FrequencyNode(key, value)
        self.cache[key] = node
        self.freq_map[1][key] = node
        self.min_freq = 1
    def _update_frequency(self, node: FrequencyNode):
        """更新节点频率"""
        old_freq = node.freq
        node.freq += 1
        # 从旧频率映射中删除
        del self.freq_map[old_freq][node.key]
        if not self.freq_map[old_freq]:
            del self.freq_map[old_freq]
            if self.min_freq == old_freq:
                self.min_freq += 1
        # 添加到新频率映射
        self.freq_map[node.freq][node.key] = node
    def _evict(self):
        """淘汰最低频率的数据"""
        if not self.freq_map:
            return
        # 获取最低频率的key列表
        keys = list(self.freq_map[self.min_freq].keys())
        if not keys:
            return
        # 淘汰最早访问的
        oldest_key = keys[0]
        oldest_node = self.freq_map[self.min_freq][oldest_key]
        for key in keys:
            node = self.freq_map[self.min_freq][key]
            if node.last_access < oldest_node.last_access:
                oldest_key = key
                oldest_node = node
        # 删除
        del self.cache[oldest_key]
        del self.freq_map[self.min_freq][oldest_key]
        if not self.freq_map[self.min_freq]:
            del self.freq_map[self.min_freq]

增强版：带权重的时间衰减

class WeightedLFUCache:
    """带时间衰减权重的LFU缓存"""
    def __init__(self, capacity: int, decay_factor: float = 0.95):
        self.capacity = capacity
        self.decay_factor = decay_factor
        self.cache = {}
        self.freq_heap = []  # 小顶堆，用于快速获取最低频率
        self.time_scale = 3600  # 时间缩放因子（秒）
    def _calculate_weight(self, node):
        """计算加权频率"""
        elapsed = time.time() - node.last_access
        time_weight = self.decay_factor ** (elapsed / self.time_scale)
        return node.freq * time_weight
    def get(self, key):
        if key not in self.cache:
            return None
        node = self.cache[key]
        node.freq += 1
        node.last_access = time.time()
        # 更新堆
        self._update_heap(key)
        return node.value
    def put(self, key, value):
        if self.capacity <= 0:
            return
        if key in self.cache:
            node = self.cache[key]
            node.value = value
            node.freq += 1
            self._update_heap(key)
            return
        # 淘汰策略
        if len(self.cache) >= self.capacity:
            self._evict()
        node = FrequencyNode(key, value)
        self.cache[key] = node
        heapq.heappush(self.freq_heap, (self._calculate_weight(node), key))
    def _update_heap(self, key):
        """更新堆中节点"""
        node = self.cache[key]
        # 重建堆（实际可使用更高效的方式）
        new_heap = []
        for weight, k in self.freq_heap:
            if k == key:
                continue
            new_heap.append((self._calculate_weight(self.cache[k]), k))
        new_heap.append((self._calculate_weight(node), key))
        heapq.heapify(new_heap)
        self.freq_heap = new_heap
    def _evict(self):
        """淘汰权重最低的"""
        while self.freq_heap:
            weight, key = heapq.heappop(self.freq_heap)
            if key in self.cache:
                del self.cache[key]
                return

分层缓存管理策略

class TieredHotCache:
    """分层热数据缓存"""
    def __init__(self, 
                 hot_capacity: int,      # 热数据容量
                 warm_capacity: int,     # 温数据容量  
                 hot_threshold: int = 10): # 热数据频率阈值
        self.hot_cache = LFUCache(hot_capacity)
        self.warm_cache = LFUCache(warm_capacity)
        self.hot_threshold = hot_threshold
    def get(self, key):
        # 优先从热数据获取
        value = self.hot_cache.get(key)
        if value is not None:
            return value
        # 从温数据获取
        value = self.warm_cache.get(key)
        if value is not None:
            # 如果频率达到阈值，升级到热数据
            if key in self.warm_cache.cache:
                node = self.warm_cache.cache[key]
                if node.freq >= self.hot_threshold:
                    self._promote(key, value)
            return value
        return None
    def put(self, key, value):
        # 默认放入温数据
        self.warm_cache.put(key, value)
    def _promote(self, key, value):
        """将数据提升到热数据层"""
        self.hot_cache.put(key, value)
        if key in self.warm_cache.cache:
            del self.warm_cache.cache[key]

使用示例

# 基础使用
hot_cache = HotCacheSystem(capacity=1000)
# 存入数据
hot_cache.set("user:1001", {"name": "张三", "score": 95})
hot_cache.set("user:1002", {"name": "李四", "score": 88})
# 访问数据（频率增加）
user = hot_cache.get("user:1001")
# 批量操作
batch_data = {
    "article:1": "内容1",
    "article:2": "内容2",
    "article:3": "内容3"
}
for key, value in batch_data.items():
    hot_cache.set(key, value)
# 获取缓存统计信息
stats = hot_cache.get_stats()
print(f"命中率: {stats['hit_rate']:.2%}")
print(f"访问次数: {stats['access_count']}")

优化建议

1. 并发安全：使用 threading.Lock 或 asyncio.Lock
2. 过期策略：结合TTL（生存时间）淘汰冷数据
3. 持久化：定期将热数据保存到磁盘
4. 自适应阈值：根据系统负载动态调整频率阈值

这个系统能有效保证高频访问的数据常驻缓存,提高系统性能。

标签： 冷数据