Python cachetools常用缓存算法汇总

embedded/2024/10/23 18:35:21/

文章目录

        • cachetools介绍
        • 缓存操作
        • cachetools 使用示例

cachetools介绍

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

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

    1. LRUCache（Least Recently Used Cache）：基于最近使用的原则，删除最久未使用的缓存项。当缓存达到最大容量时，将删除最久未使用的缓存项。

    2. LFUCache（Least Frequently Used Cache）：基于最近使用频率的原则，删除使用频率最低的缓存项。当缓存达到最大容量时，将删除使用频率最低的缓存项。

    3. FIFOCache（First In, First Out Cache）：按照缓存项的插入顺序进行删除，最先插入的缓存项将首先被删除。

    4. RRCache（Random Replacement Cache）：随机删除缓存项，没有特定的策略。

这些缓存策略都可以在 Cachetools 中使用，并可以通过设置缓存的最大容量来控制缓存的大小。

python">import cachetools# 创建 LRU 缓存
lru_cache = cachetools.LRUCache(maxsize=100)# 创建 MRU 缓存
mru_cache = cachetools.MRUCache(maxsize=100)# 创建 RR 缓存
rr_cache = cachetools.RRCache(maxsize=100)# 创建 FIFO 缓存
fifo_cache = cachetools.FIFOCache(maxsize=100)

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

缓存操作

缓存对象支持类似字典的操作，例如：添加、获取、删除和更新缓存项。

python"># 类似于字典操作# 添加缓存项
lru_cache["key"] = "value"# 获取缓存项
value = lru_cache.get("key", "default_value")
print(lru_cache)
# 删除缓存项
if "key" in lru_cache:del lru_cache["key"]# 更新缓存项
lru_cache["key"] = "new_value"
print(lru_cache)

python">LRUCache({'key': 'value'}, maxsize=100, currsize=1)
LRUCache({'key': 'new_value'}, maxsize=100, currsize=1)
设置数据生存时间（TTL）

cachetools 还支持为缓存项设置生存时间（TTL）。当缓存项的生存时间到期后，该项将被自动移除。

python">import cachetools
import time# 创建一个带 TTL 的缓存对象
ttl_cache = cachetools.TTLCache(maxsize=100, ttl=60)# 添加缓存项
ttl_cache["key"] = "value"
print(ttl_cache)# 等待一段时间，让缓存项过期
time.sleep(61)# 此时缓存项已过期，尝试获取时将返回默认值
value = ttl_cache.get("key", "default_value")
print(value)

当为ttl_cache添加缓存项之后，可以看到 TTLCache类型缓存添加成功，当过去61s之后，缓存项已过期，尝试获取时返回的是默认值default_value。

python">TTLCache({'key': 'value'}, maxsize=100, currsize=1)
default_value
自定义缓存策略

cachetools允许自定义缓存策略。要实现一个自定义的缓存策略，需要继承 cachetools.Cache 类，并实现相应的方法。例如，实现一个简单的大小有限制的缓存：

python">import cachetoolsclass 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)
custom_cacheSizeLimitedCache({}, maxsize=100, currsize=0)
缓存装饰器

cachetools还提供了一些缓存装饰器，可以方便地将缓存应用于函数或方法。

python">import cachetools
import cachetools.func
import requests # 使用 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"})
缓存清理

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

python">import cachetools# 创建 LRU 缓存
lru_cache = cachetools.LRUCache(maxsize=100)
lru_cache["name"] = "Abel"
lru_cache["age"] = 33
lru_cache["job"] = "student"
print(lru_cache)# 移除最近最少使用的缓存项
lru_cache.popitem()
print(lru_cache)# 手动清空缓存
lru_cache.clear()
print(lru_cache)

lru_cache创建缓存之后依次添加了3个缓存项，当使用popitem()函数移除最近最少使用的一条缓存项之后，lru_cache只剩余其他两个缓存项，最后使用clear()函数清空缓存之后，lru_cache显示为空。

python">LRUCache({'name': 'Abel', 'age': 33, 'job': 'student'}, maxsize=100, currsize=3)
LRUCache({'age': 33, 'job': 'student'}, maxsize=100, currsize=2)
LRUCache({}, maxsize=100, currsize=0)

python">import cachetools# 创建 LRU 缓存
lru_cache = cachetools.LRUCache(maxsize=100)# 向lru_cache添加缓存项
lru_cache["name"] = "Abel"
lru_cache["age"] = 33
lru_cache["job"] = "student"
print(lru_cache)# 查看缓存项
print(lru_cache.get("name"))# 移除最近最少使用的缓存项
lru_cache.popitem()
print(lru_cache)# 手动清空缓存
lru_cache.clear()
print(lru_cache)

在添加3条缓存项之后，如果查看第一条缓存项，则这条缓存项被使用过，在调用popitem()函数进行移除时，会从下一条未被使用过的缓存项开始，找到最近的一条进行删除。

python">LRUCache({'name': 'Abel', 'age': 33, 'job': 'student'}, maxsize=100, currsize=3)
Abel
LRUCache({'name': 'Abel', 'job': 'student'}, maxsize=100, currsize=2)
LRUCache({}, maxsize=100, currsize=0)
cachetools 超过缓存数量maxsize

设置缓存量为10，添加11个缓存项，对LRUCache缓存策略，则会删除最近未被使用的一条。

python">import cachetools# 创建 LRU 缓存
lru_cache = cachetools.LRUCache(maxsize=10)for index in range(11):lru_cache[index] = 'cache'+str(index)print(lru_cache)

python">LRUCache({'key': 'value'}, maxsize=100, currsize=1)
LRUCache({'key': 'new_value'}, maxsize=100, currsize=1)
TTLCache({'key': 'value'}, maxsize=100, currsize=1)
default_value
SizeLimitedCache({}, maxsize=100, currsize=0)
---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
Cell In[11], line 1411     return data13 # 使用缓存的函数
---> 14 data = get_data_from_api("https://api.example.com/data", {"param1": "value1", "param2": "value2"})File e:\python3.10\lib\site-packages\cachetools\__init__.py:696, in cached.<locals>.decorator.<locals>.wrapper(*args, **kwargs)694 try:695     with lock:
--> 696         result = cache[k]697         hits += 1698         return resultFile e:\python3.10\lib\site-packages\cachetools\__init__.py:410, in TTLCache.__getitem__(self, key, cache_getitem)408 def __getitem__(self, key, cache_getitem=Cache.__getitem__):409     try:
--> 410         link = self.__getlink(key)411     except KeyError:412         expired = FalseFile e:\python3.10\lib\site-packages\cachetools\__init__.py:497, in TTLCache.__getlink(self, key)496 def __getlink(self, key):
--> 497     value = self.__links[key]498     self.__links.move_to_end(key)
...18 if hashvalue is None:
---> 19     self.__hashvalue = hashvalue = hash(self)20 return hashvalueTypeError: unhashable type: 'dict'
Output is truncated. View as a scrollable element or open in a text editor. Adjust cell output settings...
LRUCache({'name': 'Abel', 'age': 33, 'job': 'student'}, maxsize=100, currsize=3)
Abel
LRUCache({'name': 'Abel', 'job': 'student'}, maxsize=100, currsize=2)
LRUCache({}, maxsize=100, currsize=0)
---------------------------------------------------------------------------
gaierror                                  Traceback (most recent call last)
File e:\python3.10\lib\site-packages\urllib3\connection.py:174, in HTTPConnection._new_conn(self)173 try:
--> 174     conn = connection.create_connection(175         (self._dns_host, self.port), self.timeout, **extra_kw176     )178 except SocketTimeout:File e:\python3.10\lib\site-packages\urllib3\util\connection.py:72, in create_connection(address, timeout, source_address, socket_options)68     return six.raise_from(69         LocationParseError(u"'%s', label empty or too long" % host), None70     )
---> 72 for res in socket.getaddrinfo(host, port, family, socket.SOCK_STREAM):73     af, socktype, proto, canonname, sa = resFile e:\python3.10\lib\socket.py:955, in getaddrinfo(host, port, family, type, proto, flags)954 addrlist = []
--> 955 for res in _socket.getaddrinfo(host, port, family, type, proto, flags):956     af, socktype, proto, canonname, sa = resgaierror: [Errno 11001] getaddrinfo failedDuring handling of the above exception, another exception occurred:
...
--> 519     raise ConnectionError(e, request=request)521 except ClosedPoolError as e:522     raise ConnectionError(e, request=request)ConnectionError: HTTPSConnectionPool(host='api.example.com', port=443): Max retries exceeded with url: /data (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x0000015BB5EBEFB0>: Failed to establish a new connection: [Errno 11001] getaddrinfo failed'))
Output is truncated. View as a scrollable element or open in a text editor. Adjust cell output settings...
---------------------------------------------------------------------------
gaierror                                  Traceback (most recent call last)
File e:\python3.10\lib\site-packages\urllib3\connection.py:174, in HTTPConnection._new_conn(self)173 try:
--> 174     conn = connection.create_connection(175         (self._dns_host, self.port), self.timeout, **extra_kw176     )178 except SocketTimeout:File e:\python3.10\lib\site-packages\urllib3\util\connection.py:72, in create_connection(address, timeout, source_address, socket_options)68     return six.raise_from(69         LocationParseError(u"'%s', label empty or too long" % host), None70     )
---> 72 for res in socket.getaddrinfo(host, port, family, socket.SOCK_STREAM):73     af, socktype, proto, canonname, sa = resFile e:\python3.10\lib\socket.py:955, in getaddrinfo(host, port, family, type, proto, flags)954 addrlist = []
--> 955 for res in _socket.getaddrinfo(host, port, family, type, proto, flags):956     af, socktype, proto, canonname, sa = resgaierror: [Errno 11001] getaddrinfo failedDuring handling of the above exception, another exception occurred:
...
--> 519     raise ConnectionError(e, request=request)521 except ClosedPoolError as e:522     raise ConnectionError(e, request=request)ConnectionError: HTTPSConnectionPool(host='api.example.com', port=443): Max retries exceeded with url: /data?param1=value1&param2=value2 (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x0000015BB65BDED0>: Failed to establish a new connection: [Errno 11001] getaddrinfo failed'))
Output is truncated. View as a scrollable element or open in a text editor. Adjust cell output settings...
LRUCache({0: 'cache0'}, maxsize=10, currsize=1)
LRUCache({0: 'cache0', 1: 'cache1'}, maxsize=10, currsize=2)
LRUCache({0: 'cache0', 1: 'cache1', 2: 'cache2'}, maxsize=10, currsize=3)
LRUCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3'}, maxsize=10, currsize=4)
LRUCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4'}, maxsize=10, currsize=5)
LRUCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5'}, maxsize=10, currsize=6)
LRUCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5', 6: 'cache6'}, maxsize=10, currsize=7)
LRUCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5', 6: 'cache6', 7: 'cache7'}, maxsize=10, currsize=8)
LRUCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5', 6: 'cache6', 7: 'cache7', 8: 'cache8'}, maxsize=10, currsize=9)
LRUCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5', 6: 'cache6', 7: 'cache7', 8: 'cache8', 9: 'cache9'}, maxsize=10, currsize=10)
LRUCache({1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5', 6: 'cache6', 7: 'cache7', 8: 'cache8', 9: 'cache9', 10: 'cache10'}, maxsize=10, currsize=10)

Random 缓存会随机删除一条记录。

python">import cachetools# 创建 RR 缓存
rr_cache = cachetools.RRCache(maxsize=10)for index in range(11):rr_cache[index] = 'cache'+str(index)print(rr_cache)

python">RRCache({0: 'cache0'}, maxsize=10, currsize=1)
RRCache({0: 'cache0', 1: 'cache1'}, maxsize=10, currsize=2)
RRCache({0: 'cache0', 1: 'cache1', 2: 'cache2'}, maxsize=10, currsize=3)
RRCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3'}, maxsize=10, currsize=4)
RRCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4'}, maxsize=10, currsize=5)
RRCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5'}, maxsize=10, currsize=6)
RRCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5', 6: 'cache6'}, maxsize=10, currsize=7)
RRCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5', 6: 'cache6', 7: 'cache7'}, maxsize=10, currsize=8)
RRCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5', 6: 'cache6', 7: 'cache7', 8: 'cache8'}, maxsize=10, currsize=9)
RRCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5', 6: 'cache6', 7: 'cache7', 8: 'cache8', 9: 'cache9'}, maxsize=10, currsize=10)
RRCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 5: 'cache5', 6: 'cache6', 7: 'cache7', 8: 'cache8', 9: 'cache9', 10: 'cache10'}, maxsize=10, currsize=10)

FIFO缓存会删除第一条存入的缓存项。

python">import cachetools# 创建 FIFO 缓存
fifo_cache = cachetools.FIFOCache(maxsize=10)for index in range(11):fifo_cache[index] = 'cache'+str(index)print(fifo_cache)

python">FIFOCache({0: 'cache0'}, maxsize=10, currsize=1)
FIFOCache({0: 'cache0', 1: 'cache1'}, maxsize=10, currsize=2)
FIFOCache({0: 'cache0', 1: 'cache1', 2: 'cache2'}, maxsize=10, currsize=3)
FIFOCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3'}, maxsize=10, currsize=4)
FIFOCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4'}, maxsize=10, currsize=5)
FIFOCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5'}, maxsize=10, currsize=6)
FIFOCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5', 6: 'cache6'}, maxsize=10, currsize=7)
FIFOCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5', 6: 'cache6', 7: 'cache7'}, maxsize=10, currsize=8)
FIFOCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5', 6: 'cache6', 7: 'cache7', 8: 'cache8'}, maxsize=10, currsize=9)
FIFOCache({0: 'cache0', 1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5', 6: 'cache6', 7: 'cache7', 8: 'cache8', 9: 'cache9'}, maxsize=10, currsize=10)
FIFOCache({1: 'cache1', 2: 'cache2', 3: 'cache3', 4: 'cache4', 5: 'cache5', 6: 'cache6', 7: 'cache7', 8: 'cache8', 9: 'cache9', 10: 'cache10'}, maxsize=10, currsize=10)
cachetools 使用示例

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

python">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")

http://www.ppmy.cn/embedded/128065.html

相关文章

The First项目报告：Puffer Finance如何破解LST/LRT赛道中心化难题？

The First项目报告：Puffer Finance如何破解LST/LRT赛道中心化难题？

10月09日The First上线了一种基于以太坊的协议&#xff0c;引入了重新质押的概念的生态协议EigenLayer。10月15日&#xff0c;一款搭建在 Eigenlayer 上原生流动性再质押平台 Puffer Finance 宣布其原生代币 PUFFER 将于 2024 年 10 月 15 日 07时&#xff08;UTC8&#xff09;…
阅读更多...
微信小程序：miniprogram-ci自动打包工具使用介绍以及支持配置环境变量、jekins打包、taro、uni-app三方工具

微信小程序：miniprogram-ci自动打包工具使用介绍以及支持配置环境变量、jekins打包、taro、uni-app三方工具

微信小程序&#xff1a;miniprogram-ci自动打包工具使用介绍以及支持配置环境变量、jekins打包、taro、uni-app三方工具 背景介绍 一直都是本地电脑运行微信开发者工具打包上传。多项目中新老版本对node版本要求不一致&#xff0c;老是切来切去。而且同一个人开发上传需要打包…
阅读更多...
mysql用户管理(user表列信息介绍,本质,管理操作),数据库的权限管理(权限列表,权限操作)

mysql用户管理(user表列信息介绍,本质,管理操作),数据库的权限管理(权限列表,权限操作)

目录 用户管理 介绍 user表 介绍 列信息 Host User *_priv authentication_string 用户管理的本质 操作 创建用户 删除用户 修改用户信息 修改密码 自己修改 root用户修改指定用户的密码 数据库的权限 权限列表 给用户授权 查看权限 回收权限 刷新权限 …
阅读更多...
idea中高级实用的调试技巧

idea中高级实用的调试技巧

1 条件断点&#xff0c;比如&#xff1a;遍历1个大List的过程中&#xff0c;想让断点停在某个特定值 /*** 1、条件断点*/Testpublic void test1() {for (int i 0; i < 10; i) {System.out.println(i);}} 2 异常断点&#xff0c;在程序中出现需要拦截的异常时&#xff0c;会…
阅读更多...
python的异常处理

python的异常处理

Python中的异常处理&#xff1a; try…except的语法结构为&#xff1a; try: 可能会抛出异常的代码 except 异常类型: 异常处理代码&#xff08;报错后执行的代码&#xff09; try…except…except的语法结构为&#xff1a; try: 可能会抛出异常的代码 except …
阅读更多...
成都睿明智科技有限公司电商服务可靠不？

成都睿明智科技有限公司电商服务可靠不？

在这个短视频风起云涌的时代&#xff0c;抖音不仅成为了人们娱乐消遣的首选平台&#xff0c;更是众多商家竞相追逐的电商新蓝海。成都睿明智科技有限公司&#xff0c;作为抖音电商服务领域的佼佼者&#xff0c;正以其独到的洞察力和专业的服务&#xff0c;助力无数品牌在这片沃…
阅读更多...
多态常见面试问题

多态常见面试问题

1、什么是多态&#xff1f; 多态&#xff08;Polymorphism&#xff09;是面向对象编程中的一个重要概念&#xff0c;它允许同一个接口表现出不同的行为。在C中&#xff0c;多态性主要通过虚函数来实现&#xff0c;分为编译时多态&#xff08;静态多态&#xff09;和运行时多态…
阅读更多...
多种方式连接和管理 Oracle 数据库详解

多种方式连接和管理 Oracle 数据库详解

连接/管理方式描述优点缺点使用场景SQL*Plus命令行工具&#xff0c;用于执行SQL、PL/SQL以及各种Oracle命令。轻量级&#xff0c;无需图形界面&#xff1b;适合脚本自动化。用户界面不友好&#xff1b;功能相对有限。开发者快速查询&#xff0c;数据库维护。SQL Developer免费的…
阅读更多...
最新文章

Copyright @ 2022~2023