Python_cachetools实现缓存过期策略

Python?cachetools实现缓存过期策略

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

安装与配置

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

pip install cachetools

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

核心概念与用法

1. 缓存类型介绍

LRUCache（Least Recently Used） ：最近最少使用缓存，会优先移除最久未被访问的项目。
TTLCache（Time-To-Live） ：设置缓存项的存活时刻，过期后自动删除。
LFUCache（Least Frequently Used） ：最少频繁使用缓存，会移除访问频率最低的项目。

2. 创建缓存对象

下面内容是创建不同类型缓存的示例：

from cachetools import LRUCache, TTLCache, LFUCache 创建 LRU 缓存，最大容量为 5lru_cache = LRUCache(maxsize=5) 创建 TTL 缓存，最大容量为 10，每项存活时刻为 60 秒ttl_cache = TTLCache(maxsize=10, ttl=60) 创建 LFU 缓存，最大容量为 3lfu_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 LRUCachecache = LRUCache(maxsize=2) 添加数据到缓存中cache[‘a’] = 1cache[‘b’] = 2print(cache) 输出: ‘a’: 1, ‘b’: 2} 添加新数据，超出容量时移除最少使用的数据cache[‘c’] = 3print(cache) 输出: ‘b’: 2, ‘c’: 3}

5. 自定义键生成函数

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

from cachetools import cached, LRUCache, keysdef 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 requestsfrom 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 Noneurl = “https://jsonplaceholder.typicode.com/todos/1″print(fetch_api_data(url)) 实际请求 APIprint(fetch_api_data(url)) 从缓存获取结局

性能测试与优化效果

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

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

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

Python cachetools 实现缓存过期策略

实现代码

from cachetools import TTLCacheimport 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”&x1f6a8; 缓存过期！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 timefrom 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”&x1f514; 回调通知: 参数 key} 的值 value} 已过期”) 应用装饰器@timed_cache(ttl=3, callback=expire_callback)def heavy_compute(x): print(f”计算中… 参数=x}”) return x 2 测试print(heavy_compute(4)) 首次计算，输出: 计算中… 16print(heavy_compute(4)) 命中缓存，无输出 → 16time.sleep(4)print(heavy_compute(4)) 过期后重新计算，触发回调 → 计算中… 16

具体要怎么做

cachetools 提供了多种灵活的缓存策略（如 LRU、TTL、LFU 等），适用于不同场景。
使用简单，可通过装饰器快速实现函数结局缓存。
在实际开发中，应根据需求选择合适的缓存类型和参数配置。

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

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

无论兄弟们可能感兴趣的文章:

Python缓存利器之cachetools库使用详解
python3自动更新缓存类的具体使用
Python实现缓存的两个简单技巧
Python实现本地缓存的几种技巧
Python使用LRU缓存策略进行缓存的技巧步骤

欢乐号