functools
模块中的 lru_cache
(Least Recently Used Cache)是 Python 中提供的一种缓存装饰器,用于缓存函数的结果。LRU Cache是一种缓存策略,它会保留最近使用的数据,而淘汰最长时间没有被使用的数据。使用 lru_cache
可以有效地提高函数的性能,特别是对于那些计算开销较大、但结果不经常改变的函数。
以下是 functools.lru_cache
的一些关键特点:
装饰器: lru_cache
被设计为一个装饰器,可以应用于函数。通过在函数定义之前使用 @lru_cache
,可以启用缓存功能。
from functools import lru_cache
@lru_cache(maxsize=None)
def my_function(arg):
# 函数体
缓存大小: 可以通过 maxsize
参数设置缓存的大小,即可以存储的元素数量。如果将 maxsize
设置为 None
(默认值),缓存的大小将不受限制。
缓存键: 默认情况下,lru_cache
使用函数的参数作为缓存的键。如果函数被调用时,参数与之前调用相同,而且缓存中有相应的结果,那么缓存中的结果会被直接返回,而不执行函数体。
自动清理: 当缓存的大小超过指定的 maxsize
时,lru_cache
将自动清理最近最少使用的元素,以腾出空间存储新的结果。
哈希函数: 缓存的键是通过将函数的参数进行哈希运算而生成的,因此函数的参数必须是可哈希的。
第二个参数typed
: 如果设置为 True
,则将不同类型的参数分别缓存。例如,对于参数 1
和 1.0
,如果 typed
是 True
,它们将被视为不同的参数。
下面是一个简单的例子:
from functools import lru_cache
@lru_cache(maxsize=None)
def get_original_number(n: int) -> int:
print(f"The function has been executed: {n}")
return n
# 第一次调用会执行函数体,并将结果缓存
result: int = get_original_number(5)
result = get_original_number(6)
result = get_original_number(7)
# 第二次调用时,直接从缓存中获取结果,不再执行函数体
cached_result = get_original_number(5)
得到运行结果:
The function has been executed: 5
The function has been executed: 6
The function has been executed: 7
from cachetools import cached, LRUCache
cache = LRUCache(maxsize=128)
@cached(cache=cache)
def my_function(arg):
# 函数体
这里的LRUCache
类的构造有两个参数:
class LRUCache(Cache):
"""Least Recently Used (LRU) cache implementation."""
def __init__(self, maxsize, getsizeof=None):
Cache.__init__(self, maxsize, getsizeof)
self.__order = collections.OrderedDict()
先看maxsize
指定最大缓存数,需要注意它的淘汰模式:
from cachetools import cached, LRUCache
my_define_cache: LRUCache = LRUCache(maxsize=3)
@cached(my_define_cache)
def my_cache_test(value: int) -> int:
print(f"The function has been executed: {value}")
return value
if __name__ == "__main__":
# 先添加三个值占满缓存
my_cache_test(value=1)
my_cache_test(value=2)
my_cache_test(value=3)
# 重复运行已有值不会重复执行 但是会改变缓存中的键值顺序
# 使得之前的3被放在末尾位置
my_cache_test(value=1)
my_cache_test(value=2)
# 添加新值 3被淘汰 故会再次运行3
my_cache_test(value=4)
my_cache_test(value=3)
现在来看getsizeof
,该参数接受一个仅接受一个参数的函数作为参数,用于测量缓存的总大小,我们以下列程序进行实验:
from cachetools import cached, LRUCache
import sys
my_lru_cache: LRUCache = LRUCache(maxsize=200, getsizeof=custom_getsizeof)
def custom_getsizeof(value: str) -> int:
return sys.getsizeof(value)
def get_cache_info() -> None:
print("Cache size: ", my_lru_cache.currsize)
print("Cache info: ", my_lru_cache, "\n")
@cached(my_lru_cache)
def my_lru_cache_test(value: str) -> str:
print(f"The function has been executed: {value}")
return value
if __name__ == "__main__":
my_lru_cache_test("2333")
get_cache_info()
my_lru_cache_test("23333")
get_cache_info()
my_lru_cache_test("233" * 20)
get_cache_info()
my_lru_cache_test("2333")
get_cache_info()
在这个程序中我定义了一个返回对象所占对象字节大小的方法(可以包含其他操作,这里仅作示例),然后指定了一个缓存的最大字节大小为200字节,得到以下运行结果(过长影响观看字符串我进行了省略):
The function has been executed: 2333
Cache size: 53
Cache info: LRUCache({('2333',): '2333'}, maxsize=200, currsize=53)
The function has been executed: 23333
Cache size: 107
Cache info: LRUCache({('2333',): '2333', ('23333',): '23333'}, maxsize=200, currsize=107)
The function has been executed: 233233233233233233233233233233233233233233233233233233233233
Cache size: 163
Cache info: LRUCache({('23333',): '23333', ('2332332332 ....',): '23323323 ....'}, maxsize=200, currsize=163)
The function has been executed: 2333
Cache size: 162
Cache info: LRUCache({('2332332332 ....',): '23323323 ....', ('2333',): '2333'}, maxsize=200, currsize=162)
当超过了最大字节数,将淘汰最早的键值对,而保留最新的键值对,需要注意的是,当超过缓存长度的数据加入时,将会被丢弃,例如:
my_lru_cache_test("233" * 200)
get_cache_info()
得到结果的部分展示:
The function has been executed: 23333
Cache size: 107
Cache info: LRUCache({('2333',): '2333', ('23333',): '23333'}, maxsize=200, currsize=107)
The function has been executed: 233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233233
Cache size: 107
Cache info: LRUCache({('2333',): '2333', ('23333',): '23333'}, maxsize=200, currsize=107)
关键特点:
Time-To-Live (TTL): TTLCache
允许为缓存中的每个条目设置一个存活时间。一旦条目超过其 TTL,它将被认为过期并从缓存中移除。
自动过期: TTLCache
通过检查条目的存活时间来自动处理过期。当一个条目超过其 TTL 时,它将被自动移除,不再可用。
缓存大小限制: 可以配置 TTLCache
的最大大小,以确保缓存不会无限增长。当达到设定的大小时,最早添加的条目将被淘汰以腾出空间。
线程安全: TTLCache
是线程安全的,这意味着它可以在多线程环境中安全地使用,而无需额外的同步。
maxsize:
128
。from cachetools import TTLCache
ttl_cache = TTLCache(maxsize=100)
ttl:
datetime.timedelta
对象。默认为300
秒(5分钟)。from cachetools import TTLCache
ttl_cache = TTLCache(maxsize=100, ttl=60) # 缓存项存活60秒
getsizeof:
None
from cachetools import TTLCache
def custom_getsizeof(key, value):
# 自定义计算缓存项大小的函数
return len(value)
ttl_cache = TTLCache(maxsize=100, ttl=60, getsizeof=custom_getsizeof)
和LRUCache的使用是大同小异的,只是需要注意生效时间的限制:
from cachetools import cached, TTLCache
from sys import getsizeof
from time import sleep
my_ttl_cache: TTLCache = TTLCache(maxsize=100, ttl=3, getsizeof=getsizeof)
def get_ttl_cache_info() -> None:
print("Cache size: ", my_ttl_cache.currsize)
print("Cache info: ", my_ttl_cache, "\n")
@cached(my_ttl_cache)
def my_ttl_cache_test(value: str) -> str:
print(f"The function has been executed: {value}")
return value
if __name__ == "__main__":
my_ttl_cache_test("233")
get_ttl_cache_info()
sleep(3)
my_ttl_cache_test("233")
get_ttl_cache_info()
得到运行结果:
The function has been executed: 233
Cache size: 52
Cache info: TTLCache({('233',): '233'}, maxsize=100, currsize=52)
The function has been executed: 233
Cache size: 52
Cache info: TTLCache({('233',): '233'}, maxsize=100, currsize=52)
The core of DiskCache is three data types intended for caching. Cache objects manage a SQLite database and filesystem directory to store key and value pairs.
from:【PyPi】https://pypi.org/project/diskcache/
diskcache
是一个用于在磁盘上进行持久化缓存的 Python 库。它允许将数据缓存到磁盘上,以便在多个运行之间共享数据,并在应用程序重新启动时保持数据的可用性。以下是 diskcache
库的一些主要特性和用法:
持久化缓存: diskcache
允许将数据缓存到磁盘上,这样即使应用程序重新启动,之前缓存的数据仍然可用。这对于需要跨应用程序生命周期保留数据的场景非常有用。
字典接口: diskcache
提供了类似字典的接口,使得使用起来非常直观。您可以使用键值对的方式存储和检索数据。
自动过期: 您可以为缓存中的条目设置过期时间,类似于 TTLCache
。过期的条目将在访问时自动删除,释放磁盘空间。
线程安全: diskcache
是线程安全的,这意味着可以在多线程环境中使用而无需额外的同步。
多格式支持: diskcache
支持多种数据格式,包括二进制、JSON 等。这使得它适用于各种数据类型的缓存需求。
以下是一个简单的示例,演示如何使用 diskcache
:
from diskcache import Cache
# 创建一个磁盘缓存
cache = Cache('./cache')
# 将数据存储到缓存中
cache['key1'] = 'value1'
# 从缓存中获取数据
result = cache.get('key1')
# 检查是否获取到值
if result is not None:
print(result)
else:
print("Key not found in cache.")
# 打印结果
print(result)
示例中,Cache
被创建为一个磁盘缓存在当前路径的cache
目录下,并且可以通过键值对的方式进行存储和检索。对于之前存在的缓存也是如此,需要注意的是:在读取缓存时,如果缓存中的数据已过期(如果设置了过期时间),diskcache
将会自动删除过期的条目。
cache_dict: dict[str, str] = {}
def cache_controler(arg: str) -> str | None:
if arg in cache_dict.values:
return cache_dict[arg]
else:
cache_dict[arg] = arg
return None
手动实现缓存时,需要维护一个字典(或其他数据结构)来存储缓存数据,并在需要时进行清理和过期处理,通常我们会像上述模块一般定义一个类,具体依靠业务逻辑来使用