Python编程-利用缓存管理实现程序提速

发布时间:2024年01月21日

Python编程-利用缓存管理实现程序提速

functools模块-lru-cache方法

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):
    # 函数体
  1. 缓存大小: 可以通过 maxsize 参数设置缓存的大小,即可以存储的元素数量。如果将 maxsize 设置为 None(默认值),缓存的大小将不受限制。

  2. 缓存键: 默认情况下,lru_cache 使用函数的参数作为缓存的键。如果函数被调用时,参数与之前调用相同,而且缓存中有相应的结果,那么缓存中的结果会被直接返回,而不执行函数体。

  3. 自动清理: 当缓存的大小超过指定的 maxsize 时,lru_cache 将自动清理最近最少使用的元素,以腾出空间存储新的结果。

  4. 哈希函数: 缓存的键是通过将函数的参数进行哈希运算而生成的,因此函数的参数必须是可哈希的。

  5. 第二个参数typed: 如果设置为 True,则将不同类型的参数分别缓存。例如,对于参数 11.0,如果 typedTrue,它们将被视为不同的参数。

下面是一个简单的例子:

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

cachetools模块-LRUCache类

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)

cachetools模块-TTLCache类

关键特点:

  1. Time-To-Live (TTL): TTLCache 允许为缓存中的每个条目设置一个存活时间。一旦条目超过其 TTL,它将被认为过期并从缓存中移除。

  2. 自动过期: TTLCache 通过检查条目的存活时间来自动处理过期。当一个条目超过其 TTL 时,它将被自动移除,不再可用。

  3. 缓存大小限制: 可以配置 TTLCache 的最大大小,以确保缓存不会无限增长。当达到设定的大小时,最早添加的条目将被淘汰以腾出空间。

  4. 线程安全: TTLCache 是线程安全的,这意味着它可以在多线程环境中安全地使用,而无需额外的同步。

  5. maxsize:

    • 表示TTL缓存的最大容量,即最多可以存储多少个键值对。当缓存达到这个容量时,将会根据TTL策略淘汰过期的键值对,默认为128
    from cachetools import TTLCache
    
    ttl_cache = TTLCache(maxsize=100)
    
  6. ttl:

    • 表示缓存项的存活时间,即多长时间后缓存项被视为过期。可以是一个整数或一个datetime.timedelta对象。默认为300秒(5分钟)。
    from cachetools import TTLCache
    
    ttl_cache = TTLCache(maxsize=100, ttl=60)  # 缓存项存活60秒
    
  7. 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)

diskcache模块-磁盘持久化缓存

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 库的一些主要特性和用法:

  1. 持久化缓存: diskcache 允许将数据缓存到磁盘上,这样即使应用程序重新启动,之前缓存的数据仍然可用。这对于需要跨应用程序生命周期保留数据的场景非常有用。

  2. 字典接口: diskcache 提供了类似字典的接口,使得使用起来非常直观。您可以使用键值对的方式存储和检索数据。

  3. 自动过期: 您可以为缓存中的条目设置过期时间,类似于 TTLCache。过期的条目将在访问时自动删除,释放磁盘空间。

  4. 线程安全: diskcache 是线程安全的,这意味着可以在多线程环境中使用而无需额外的同步。

  5. 多格式支持: 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

手动实现缓存时,需要维护一个字典(或其他数据结构)来存储缓存数据,并在需要时进行清理和过期处理,通常我们会像上述模块一般定义一个类,具体依靠业务逻辑来使用

文章来源:https://blog.csdn.net/m0_74220316/article/details/135730105
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。