Python cachetools常用缓存算法汇总

server/2024/10/17 23:15:32/

文章目录

        • 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/server/131289.html

相关文章

异步请求与CGI开发：深入理解与实践

异步请求与CGI开发：深入理解与实践

文章目录 异步请求与CGI开发&#xff1a;深入理解与实践1. 异步请求&#xff08;JavaScript 和 Fetch API&#xff09;1.1 异步请求的概述1.2 Fetch API**GET 请求&#xff1a;****POST 请求&#xff1a;****PUT 请求&#xff1a;****DELETE 请求&#xff1a;** 1.3 XMLHttpReq…
阅读更多...
dbt doc 生成文档命令示例应用

dbt doc 生成文档命令示例应用

DBT提供了强大的命令行工具&#xff0c;它使数据分析师和工程师能够更有效地转换仓库中的数据。dbt的一个关键特性是能够为数据模型生成文档&#xff0c;这就是dbt docs命令发挥作用的地方。本教程将指导您完成使用dbt生成和提供项目文档的过程。 dbt doc 命令 dbt docs命令有…
阅读更多...
模板方法模式、策略模式(C++)

模板方法模式、策略模式(C++)

模板方法模式&#xff1a; 定义&#xff1a;定义一个操作算法的框架&#xff0c;实现步骤延迟到子类中去实现 策略模式&#xff1a; 定义&#xff1a;定义一系列的算法&#xff0c;把它们一个个封装起来&#xff0c;并且使它们可相互替换。该模式使得算法可独立于使用它的客户…
阅读更多...
macOS Sequoia 15.0.1

macOS Sequoia 15.0.1

macOS Sequoia 推出了一系列新功能&#xff0c;可助你在 Mac 上提高生产力和创造力。通过最新连续互通功能 iPhone 镜像&#xff0c;你可以在 Mac 上访问整个 iPhone。轻松平铺窗口快速打造理想工作空间&#xff0c;还可查看通过演讲者前置演示时即将共享的内容。经过重大更新的…
阅读更多...
使用 iperf3 工具测试TCP/UDP吞吐量

使用 iperf3 工具测试TCP/UDP吞吐量

测试目标 - 测试网络的 TCP 和 UDP 吞吐量性能&#xff0c;包括不同并发连接数和目标带宽条件下的表现。 测试环境 - **测试工具**: iperf3 - **固定 IP 地址**: - 服务器 IP: 192.168.1.10 - 客户端 IP: 192.168.1.20 - **端口号**: 5201 测试准备 1. **安装 iperf3**&a…
阅读更多...
Python网络爬虫技术

Python网络爬虫技术

Python网络爬虫技术详解 引言 网络爬虫&#xff08;Web Crawler&#xff09;&#xff0c;又称网络蜘蛛&#xff08;Web Spider&#xff09;或网络机器人&#xff08;Web Robot&#xff09;&#xff0c;是一种按照一定规则自动抓取互联网信息的程序或脚本。它们通过遍历网页链…
阅读更多...
安全工具 | 搭建带有 Web 仪表板的Interact.sh

安全工具 | 搭建带有 Web 仪表板的Interact.sh

介绍 Interactsh 是一个用于检测带外交互的开源工具。它是一种旨在检测导致外部交互的漏洞的工具。本文将主要介绍在子域上设置私有 Interact.sh 服务器以及部署其 Web 应用程序。只需一个 AWS EC2 或 VPS 实例和一个域。 要求 •具有静态IP的AWS EC2 / VPS •拥有自己的域…
阅读更多...
QML tableView设置role为index的问题

QML tableView设置role为index的问题

项目中&#xff0c;需要将一个tableview的第一列表示为索引&#xff0c;且索引需从1开始。于是编写代码如下&#xff1a; 下面展示一些 内联代码片。 ListModel {id: com_model;} GE_TableView {id: tableview;height: 510;anchors.top: root.top;anchors.left: root.left;anc…
阅读更多...
最新文章

Copyright @ 2022~2023