12.cachetools 库

Pupper大约 5 分钟学习笔记Python基础

1. cachetools 库简介以及详细使用

1.1 定义

cachetools: 是一个 Python 第三方库，提供了多种缓存算法的实现。 缓存: 是一种用于临时存储计算结果的技术，以避免在后续计算中重复执行相同的计算。使用缓存可以提高应用程序的性能和响应速度。

1.2 多种缓存策略

cachetools 提供了以下常见的缓存策略：

LRU (Least Recently Used): 这种策略会移除最近最少使用的条目。
- 当需要为新的条目腾出空间时，最近最少使用的条目将被移除。这是一种常见的缓存策略，适用于大多数场景。
MRU (Most Recently Used): 这种策略会移除最近最常使用的条目。
- 当需要为新的条目腾出空间时，最近最常使用的条目将被移除。这种策略适用于一些特殊场景，例如当最近访问的数据不太可能再次访问时。
RR (Random Replacement): 这种策略会随机移除一个条目。
- 当需要为新的条目腾出空间时，随机选择一个条目进行移除。这种策略实现简单，但性能相对较差。
FIFO (First-In-First-Out): 这种策略会按照条目添加到缓存的顺序进行移除。
- 先添加的条目先被移除。这种策略适用于一些特定场景，例如缓存的数据具有固定生命周期。

import cachetools

# 创建 LRU 缓存
lru_cache = cachetools.LRUCache(maxsize=100)

# 创建 MRU 缓存
mru_cache = cachetools.MostRecentlyUsed(maxsize=100)

# 创建 RR 缓存
rr_cache = cachetools.Random(maxsize=100)

# 创建 FIFO 缓存
fifo_cache = cachetools.FIFO(maxsize=100)

maxsize参数代表的是缓存中可以存储的 最大条目数量，而不是字符数。

注意

maxsize参数代表的是缓存中可以存储的 最大条目数量，而不是字符数。
- 在这个例子中，maxsize=10000 表示缓存中最多可以存储 10000 个不同的键值对。
通常，设置一个较大的 maxsize 可以提高缓存命中率，从而提高程序性能，但同时也会增加内存消耗。因此，需要权衡程序性能和内存使用之间的关系。
maxsize 参数最好设置为 2 的幂次方，这样可以在内部哈希表扩展时更加高效。例如，可以设置为 1024、2048、4096 等。

1.3 缓存操作: 缓存对象支持类似字典的操作

# 类似于字典操作

# 添加缓存项
lru_cache["key"] = "value"

# 获取缓存项
value = lru_cache.get("key", "default_value")

# 删除缓存项
if "key" in lru_cache:
    del lru_cache["key"]

# 更新缓存项
lru_cache["key"] = "new_value"

1.4 设置数据生存时间(TTL)

TTL (Time-To-Live): 是指缓存中数据的生存时间，超过这个时间后数据将被自动删除。

import cachetools
import time

# 创建一个带 TTL 的缓存对象
ttl_cache = cachetools.TTLCache(maxsize=100, ttl=60)

# 添加缓存项
ttl_cache["key"] = "value"

# 等待一段时间，让缓存项过期
time.sleep(61)

# 此时缓存项已过期，尝试获取时将返回默认值
value = ttl_cache.get("key", "default_value")

1.5 自定义缓存策略

cachetools 还支持自定义缓存策略，可以通过继承 cachetools.Cache 类并实现 __getitem__、__setitem__、__delitem__ 等方法来实现自定义缓存策略。

import cachetools

class SizeLimitedCache(cachetools.Cache):
    def __init__(self, maxsize):
        super().__init__(maxsize=maxsize)

    def __getitem__(self, key, cache_getitem=dict.__getitem__):
        return cache_getitem(self, key)

    def __setitem__(self, key, value, cache_setitem=dict.__setitem__):
        if len(self) >= self.maxsize:
            self.popitem(last=False)  # 删除第一个缓存项
        cache_setitem(self, key, value)

# 使用自定义缓存策略
custom_cache = SizeLimitedCache(maxsize=100)

1.6 缓存装饰器

cachetools 还提供了一个装饰器 cachetools.cached，可以用于缓存函数的返回值。

import cachetools

# 使用 LRU 缓存装饰函数
@cachetools.func.ttl_cache(maxsize=100, ttl=60)
def get_data_from_api(api_url, params):
    response = requests.get(api_url, params=params)
    response.raise_for_status()
    data = response.json()
    return data

# 使用缓存的函数
data = get_data_from_api("https://api.example.com/data", {"param1": "value1", "param2": "value2"})

1.7 缓存清理

cachetools 提供了一些方法, 可以手动清理缓存

import cachetools

# 创建 LRU 缓存
lru_cache = cachetools.LRUCache(maxsize=100)

# 手动清空缓存
lru_cache.clear()

# 移除所有过期缓存项
lru_cache.expire()

# 移除最近最少使用的缓存项
lru_cache.popitem(last=False)

2. cachetools 使用示例

在这个示例中，我们使用 cachetools.LRUCache 创建一个 LRU 缓存。当我们调用 get_data_from_api() 函数时，会先检查缓存中是否有数据。如果缓存中有数据，就直接返回缓存的数据，避免了重复请求接口，提高了程序性能。

import requests
import cachetools

# 创建一个 LRU 缓存，最大容量为 100
cache = cachetools.LRUCache(maxsize=100)

def get_data_from_api(url):
    if url in cache:
        return cache[url]  # 如果数据已经在缓存中，直接返回缓存的数据

    response = requests.get(url)
    response.raise_for_status()
    data = response.json()

    cache[url] = data  # 将数据存储在缓存中
    return data

# 使用缓存的函数
data = get_data_from_api("https://api.example.com/data")

3. 错误总结

3.1 TypeError: unhashable type: 'dict'

错误来源 ：我想要直接将一个字典，作为 cachetools 缓存的键（key），这样是会报错的。

这个错误是因为字典（dict）类型在 Python 中是不可哈希（unhashable）的，因此不能直接将字典作为 cachetools 缓存的键（key）。要解决这个问题，可以将字典转换为一个可哈希的类型，比如字符串（str）或元组（tuple）。

以下是一个示例，将字典转换为字符串作为键：

import requests
import cachetools
import json

cache = cachetools.LRUCache(maxsize=100)

def get_data_from_api(api_url, params):
    # 将字典转换为字符串，并确保key的顺序一致，避免相同内容的字典生成不同的字符串
    cache_key = json.dumps(params, sort_keys=True)

    if cache_key in cache:
        return cache[cache_key]

    response = requests.get(api_url, params=params)
    response.raise_for_status()
    data = response.json()

    cache[cache_key] = data
    return data

# 使用缓存的函数
params = {"param1": "value1", "param2": "value2"}
data = get_data_from_api("https://api.example.com/data", params)

在这个示例中，我们将 params 字典转换为一个字符串，然后将字符串作为缓存的键。当调用接口获取数据时，Python 会检查缓存中是否有数据。如果缓存中有数据，就直接返回缓存的数据，避免了重复请求接口，提高了程序性能。