python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python cachetools缓存

Python cachetools实现缓存过期策略

作者：程序员小jobleap

cachetools 是一个功能强大的 Python 库,用于实现多种缓存策略,帮助开发者优化程序性能,下面小编就来和大家详细讲讲cachetools的原理与应用吧

cachetools 是一个功能强大的 Python 库，用于实现多种缓存策略（如 LRU、TTL、LFU 等），帮助开发者优化程序性能。无论是频繁调用高开销函数，还是减少重复计算，cachetools 都能显著提升效率。以下是详细的基础知识点讲解及丰富的代码案例。

安装与配置

在开始使用 cachetools 之前，需要通过 pip 安装：

pip install cachetools

安装完成后，即可直接使用，无需复杂配置。

核心概念与用法

1. 缓存类型介绍

2. 创建缓存对象

以下是创建不同类型缓存的示例：

from cachetools import LRUCache, TTLCache, LFUCache

# 创建 LRU 缓存，最大容量为 5
lru_cache = LRUCache(maxsize=5)

# 创建 TTL 缓存，最大容量为 10，每项存活时间为 60 秒
ttl_cache = TTLCache(maxsize=10, ttl=60)

# 创建 LFU 缓存，最大容量为 3
lfu_cache = LFUCache(maxsize=3)

3. 函数结果缓存

通过装饰器将函数结果缓存起来，避免重复计算：

from cachetools import cached, TTLCache

# 创建一个 TTL 缓存
cache = TTLCache(maxsize=3, ttl=10)

@cached(cache)
def expensive_computation(x):
    print(f"Computing {x}...")
    return x * x

# 第一次调用会计算并缓存
print(expensive_computation(4))  # 输出: 16
# 第二次调用直接从缓存获取
print(expensive_computation(4))  # 输出: 16（无计算）

注意：当缓存项过期或超出容量时，将触发重新计算。

4. LRU 缓存示例

LRU 缓存会自动移除最近最少使用的项目：

from cachetools import LRUCache

cache = LRUCache(maxsize=2)

# 添加数据到缓存中
cache['a'] = 1
cache['b'] = 2

print(cache)  # 输出: {'a': 1, 'b': 2}

# 添加新数据，超出容量时移除最少使用的数据
cache['c'] = 3
print(cache)  # 输出: {'b': 2, 'c': 3}

5. 自定义键生成函数

可以通过自定义键函数处理复杂参数，如忽略某些参数或处理不可哈希对象：

from cachetools import cached, LRUCache, keys

def custom_key(*args, **kwargs):
    return keys.hashkey(*args) + tuple(sorted(kwargs.items()))

cache = LRUCache(maxsize=128)

@cached(cache, key=custom_key)
def fetch_data(x, y, z=None):
    print(f"Fetching data for {x}, {y}, {z}")
    return x + y + (z or 0)

print(fetch_data(1, 2, z=3))  # 实际调用
print(fetch_data(1, 2, z=3))  # 从缓存中获取

6. 完整案例：API 数据缓存

以下示例展示如何利用 TTLCache 缓存 API 请求结果，避免重复网络请求：

import requests
from cachetools import cached, TTLCache

# 创建一个 TTL 缓存，最大容量为 5，每项生存时间为 30 秒
api_cache = TTLCache(maxsize=5, ttl=30)

@cached(api_cache)
def fetch_api_data(url):
    print(f"Fetching data from {url}")
    response = requests.get(url)
    return response.json() if response.status_code == 200 else None

url = "https://jsonplaceholder.typicode.com/todos/1"
print(fetch_api_data(url))   # 实际请求 API
print(fetch_api_data(url))   # 从缓存获取结果

性能测试与优化效果

以下是 cachetools 在实际应用中的性能提升效果：

场景无缓存耗时使用 cachetools 耗时提升比例
Fibonacci 数列计算~2 秒~0.01 秒~200 倍
API 数据请求（10 次相同）~10 秒~1 秒~10 倍

通过减少重复计算和网络请求，cachetools 可显著节省资源。

Python cachetools 实现缓存过期策略

实现代码

from cachetools import TTLCache
import time
 
class CallbackTTLCache(TTLCache):
    """支持过期回调的 TTL 缓存"""
    def __init__(self, maxsize, ttl, callback=None):
        super().__init__(maxsize, ttl)
        self.callback = callback  # 过期回调函数
        self._pending_callbacks = {}  # 跟踪需要回调的项目
 
    def __setitem__(self, key, value, **kwargs):
        super().__setitem__(key, value, **kwargs)
        if self.callback:
            self._pending_callbacks[key] = value
 
    def __getitem__(self, key):
        try:
            return super().__getitem__(key)
        except KeyError:
            if key in self._pending_callbacks:
                value = self._pending_callbacks.pop(key)
                if self.callback:
                    self.callback(key, value)
            raise
 
    def popitem(self):
        """淘汰条目时触发回调"""
        try:
            key, value = super().popitem()
            if key in self._pending_callbacks:
                self._pending_callbacks.pop(key)
            if self.callback:
                self.callback(key, value)
            return key, value
        except KeyError:
            print("缓存已经为空")
            return None, None
 
    def clear_expired(self):
        """清理所有过期的项目并触发回调"""
        current_time = time.time()
        expired_keys = []
        
        # 检查所有键是否过期
        for key in list(self._pending_callbacks):
            try:
                # 尝试获取值，如果过期会抛出KeyError
                _ = self[key]
            except KeyError:
                # 如果抛出KeyError，说明已过期
                expired_keys.append(key)
        
        # 触发过期回调
        for key in expired_keys:
            if key in self._pending_callbacks:
                value = self._pending_callbacks.pop(key)
                if self.callback:
                    self.callback(key, value)
 
    def clear(self):
        """清空缓存并触发所有回调"""
        for key, value in list(self._pending_callbacks.items()):
            if self.callback:
                self.callback(key, value)
        self._pending_callbacks.clear()
        super().clear()
 
# 示例：定义过期回调函数
def on_expired(key, value):
    print(f"🚨 缓存过期！Key: {key}, Value: {value} 已清除")
 
# 初始化缓存（最大100条，有效期5秒，绑定回调）
cache = CallbackTTLCache(maxsize=100, ttl=5, callback=on_expired)
 
# 测试用例
def run_tests():
    print("=== 开始测试 ===")
    
    # 测试1：添加并立即访问缓存
    print("\n测试1：添加并立即访问缓存")
    cache["user_101"] = "Alice"
    cache["user_102"] = "Bob"
    print("初始缓存内容:", dict(cache.items()))
    
    # 测试2：等待部分时间后访问
    print("\n测试2：等待3秒后访问缓存")
    time.sleep(3)
    print("访问user_101:", cache.get("user_101", "已过期"))
    print("3秒后缓存内容:", dict(cache.items()))
    
    # 测试3：等待过期并尝试访问
    print("\n测试3：等待剩余时间直到过期")
    time.sleep(3)  # 再等3秒，总共6秒，确保过期
    print("尝试访问user_101:", cache.get("user_101", "已过期"))
    print("6秒后（过期）缓存内容:", dict(cache.items()))
    
    # 测试4：手动清理过期内容
    print("\n测试4：清理过期内容")
    cache.clear_expired()
    print("清理后缓存内容:", dict(cache.items()))
    
    # 测试5：尝试对空缓存调用popitem
    print("\n测试5：对空缓存调用popitem")
    key, value = cache.popitem()
    print(f"popitem结果 - Key: {key}, Value: {value}")
    
    print("\n=== 测试完成 ===")
 
if __name__ == "__main__":
    run_tests()
import time
from functools import wraps
 
def timed_cache(ttl=60, callback=None):
    """支持过期时间和回调的缓存装饰器"""
    cache = {}
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            key = (args, frozenset(kwargs.items()))
            current_time = time.time()
            
            # 检查缓存是否命中且未过期
            if key in cache:
                result, expire_time = cache[key]
                if current_time < expire_time:
                    return result
                elif callback:  # 触发过期回调
                    callback(key, result)
            
            # 重新计算并缓存结果
            result = func(*args, **kwargs)
            cache[key] = (result, current_time + ttl)
            return result
        return wrapper
    return decorator
 
# 定义回调函数
def expire_callback(key, value):
    print(f"🔔 回调通知: 参数 {key} 的值 {value} 已过期")
 
# 应用装饰器
@timed_cache(ttl=3, callback=expire_callback)
def heavy_compute(x):
    print(f"计算中... 参数={x}")
    return x ** 2
 
# 测试
print(heavy_compute(4))  # 首次计算，输出: 计算中... 16
print(heavy_compute(4))  # 命中缓存，无输出 → 16
time.sleep(4)
print(heavy_compute(4))  # 过期后重新计算，触发回调 → 计算中... 16

总结与建议

推荐在 Web 开发、机器学习、大数据处理等需要高效数据访问的场景中使用 cachetools。

到此这篇关于Python cachetools实现缓存过期策略的文章就介绍到这了,更多相关Python cachetools缓存内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:
阅读全文