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

  • 开发
  • 29

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，则将不同类型的参数分别缓存。例如，对于参数 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

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

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

阅读全部

相关推荐

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

    2024-01-25 08:14:02       30 阅读

  3. 利用Python编程提取身份证的信息

    2024-01-25 08:14:02       13 阅读

  5. 利用ArrayList类编写客户信息管理程序，利用Comparer接口实现自定义Sort(IComparer comparer)排序

    2024-01-25 08:14:02       21 阅读

  8. Python编程-使用logging管理程序日志

    2024-01-25 08:14:02       14 阅读

  9. 如何利用ChatGPT来提高编程效率

    2024-01-25 08:14:02       41 阅读

  10. python企业编码管理的程序（附源码）

    2024-01-25 08:14:02       20 阅读

最近更新

  4. TCP协议是安全的吗?

    2024-01-25 08:14:02       16 阅读

  12. 阿里云服务器执行yum，一直下载docker-ce-stable失败

    2024-01-25 08:14:02       16 阅读

  13. 【Python教程】压缩PDF文件大小

    2024-01-25 08:14:02       15 阅读

  18. 通过文章id递归查询所有评论（xml）

    2024-01-25 08:14:02       18 阅读

热门阅读

  5. 设计模式四(适配器模式)

    2024-01-25 08:14:02       37 阅读

  11. Azure VM安装docker

    2024-01-25 08:14:02       31 阅读

  12. 清理Docker环境

    2024-01-25 08:14:02       30 阅读

  17. MySQL（七）MySQL和Oracle、PostgreSQL的区别

    2024-01-25 08:14:02       31 阅读

  20. Property ‘sqlSessionFactory‘ or ‘sqlSessionTemplate‘ are required

    2024-01-25 08:14:02       34 阅读

  22. 【重点问题】攻击面发现及管理

    2024-01-25 08:14:02       32 阅读

  24. CentOS 7安装Mysql+Mycat

    2024-01-25 08:14:02       33 阅读

  26. node.js npm 版本不匹配问题

    2024-01-25 08:14:02       32 阅读

  27. 在direct3D中，透明度处理和D2D1_ALPHA_MODE_PREMULTIPLIED含义？

    2024-01-25 08:14:02       29 阅读

  28. 模型部署flask学习篇（二）---- flask用户登录&用户管理

    2024-01-25 08:14:02       30 阅读

  30. pdf 转html 在线预览和查询

    2024-01-25 08:14:02       35 阅读