当前位置：首页 > news >正文

基于Python实现的缓存淘汰替换策略算法，该算法将缓存分区

news 来源：原创 2025/4/30 12:30:00

以下是一个基于Python实现的缓存淘汰替换策略算法，该算法将缓存分区，并根据不同分区的优先级进行淘汰，同时会自适应地调整缓存汰换的时机和力度，还会与GPTCache自带的LRU和FIFO策略进行对比。

import time
from gptcache.manager import CacheBase, VectorBase, get_data_manager
from gptcache.processor.pre import get_prompt
from gptcache.adapter.api import init_similar_cache
from gptcache.strategy import LRUCacheStrategy, FIFOCacheStrategy

# 定义缓存类
class CustomCache:
    def __init__(self, capacity):
        self.capacity = capacity
        self.high_priority = {}
        self.hot_window = {}
        self.medium_priority = {}
        self.other_cache = {}
        self.total_count = 0

    def add(self, key, value, category):
        # 初始化缓存项字段
        cache_item = {
            'value': value,
            'freshness': 1.0,
            'activity': 1,
            'timestamp': time.time(),
            'eviction_rate': 0.0,
            'category': category
        }
        self.total_count += 1

        # 根据类别分配到不同分区
        if category == 'high':
            self.high_priority[key] = cache_item
        elif category == 'hot':
            self.hot_window[key] = cache_item
        elif category == 'medium':
            self.medium_priority[key] = cache_item
        else:
            self.other_cache[key] = cache_item

        # 检查是否需要淘汰
        if self.total_count > self.capacity:
            self.evict()

    def get(self, key):
        for cache in [self.high_priority, self.hot_window, self.medium_priority, self.other_cache]:
            if key in cache:
                cache[key]['activity'] += 1
                cache[key]['timestamp'] = time.time()
                cache[key]['freshness'] = 1.0
                return cache[key]['value']
        return None

    def evict(self):
        # 自适应调整淘汰时机和力度
        # 这里简单模拟根据当前时间进行调整，实际可根据季节性或周期性数据调整
        current_time = time.time()
        if current_time % 3600 < 1800:  # 假设每小时前半段淘汰力度大
            eviction_count = 2
        else:
            eviction_count = 1

        while self.total_count > self.capacity and eviction_count > 0:
            if self.other_cache:
                key = next(iter(self.other_cache))
                del self.other_cache[key]
            elif self.medium_priority:
                key = next(iter(self.medium_priority))
                del self.medium_priority[key]
            elif self.hot_window:
                # 淘汰热点窗口中活跃度最低的项
                min_activity_key = min(self.hot_window, key=lambda k: self.hot_window[k]['activity'])
                del self.hot_window[min_activity_key]
            elif self.high_priority:
                # 淘汰高优先级中最旧的项
                oldest_key = min(self.high_priority, key=lambda k: self.high_priority[k]['timestamp'])
                del self.high_priority[oldest_key]
            self.total_count -= 1
            eviction_count -= 1

# 对比测试函数
def compare_strategies():
    capacity = 10
    # 自定义策略
    custom_cache = CustomCache(capacity)
    # LRU策略
    lru_strategy = LRUCacheStrategy(capacity)
    lru_cache_base = CacheBase('sqlite')
    lru_vector_base = VectorBase('faiss', dimension=128)
    lru_data_manager = get_data_manager(lru_cache_base, lru_vector_base, lru_strategy)
    init_similar_cache(data_manager=lru_data_manager, pre_embedding_func=get_prompt)
    # FIFO策略
    fifo_strategy = FIFOCacheStrategy(capacity)
    fifo_cache_base = CacheBase('sqlite')
    fifo_vector_base = VectorBase('faiss', dimension=128)
    fifo_data_manager = get_data_manager(fifo_cache_base, fifo_vector_base, fifo_strategy)
    init_similar_cache(data_manager=fifo_data_manager, pre_embedding_func=get_prompt)

    # 模拟数据访问
    data = [(i, f"value_{i}", 'high' if i < 3 else 'medium' if i < 6 else 'hot' if i < 8 else 'other') for i in range(20)]
    custom_hits = 0
    lru_hits = 0
    fifo_hits = 0

    for key, value, category in data:
        # 自定义策略
        if custom_cache.get(key) is None:
            custom_cache.add(key, value, category)
        else:
            custom_hits += 1
        # LRU策略
        if lru_data_manager.get_data(key, None) is None:
            lru_data_manager.save(key, value, None, None)
        else:
            lru_hits += 1
        # FIFO策略
        if fifo_data_manager.get_data(key, None) is None:
            fifo_data_manager.save(key, value, None, None)
        else:
            fifo_hits += 1

    print(f"Custom Cache Hits: {custom_hits}")
    print(f"LRU Cache Hits: {lru_hits}")
    print(f"FIFO Cache Hits: {fifo_hits}")

if __name__ == "__main__":
    compare_strategies()

代码说明：

CustomCache类：
- __init__：初始化缓存分区和容量。
- add：将缓存项添加到相应的分区，并在需要时调用evict方法进行淘汰。
- get：根据键查找缓存项，并更新其活跃度、时间戳和新鲜度。
- evict：根据分区优先级进行淘汰，同时自适应地调整淘汰的时机和力度。
compare_strategies函数：
- 初始化自定义缓存、LRU缓存和FIFO缓存。
- 模拟数据访问，记录每个缓存的命中次数。
- 输出每个缓存的命中次数。