跳到主要内容

Python 函数缓存

什么是函数缓存？​

函数缓存是一种优化技术，它可以存储函数的调用结果，当函数使用相同的参数再次调用时，直接返回缓存的结果，而不必重新计算。这种技术特别适用于：

  1. 计算密集型函数
  2. 具有相同输入多次调用的函数
  3. 递归函数，如斐波那契数列计算

函数缓存可以显著提高程序性能，特别是在重复计算相同结果的场景下。

Python 中的内置缓存装饰器​

从Python 3.2开始，functools模块提供了lru_cache装饰器，它实现了一个基于LRU(Least Recently Used，最近最少使用)策略的缓存机制。

使用functools.lru_cache​

import functools
import time

# 使用lru_cache装饰器
@functools.lru_cache(maxsize=128)
def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

# 测试函数执行时间
def measure_time(func, *args):
    start = time.time()
    result = func(*args)
    end = time.time()
    print(f"执行时间: {end - start:.6f}秒")
    return result

# 不使用缓存的斐波那契函数
def fibonacci_no_cache(n):
    if n <= 1:
        return n
    return fibonacci_no_cache(n-1) + fibonacci_no_cache(n-2)

# 比较有缓存和无缓存的性能差异
print("计算斐波那契数列第35项:")
print("使用缓存:")
result_cached = measure_time(fibonacci, 35)
print(f"结果: {result_cached}")

print("\n不使用缓存:")
result_no_cache = measure_time(fibonacci_no_cache, 35)
print(f"结果: {result_no_cache}")

输出结果可能如下：

计算斐波那契数列第35项:
使用缓存:
执行时间: 0.000121秒
结果: 9227465

不使用缓存:
执行时间: 4.621735秒
结果: 9227465

lru_cache参数说明​

lru_cache装饰器接受以下参数：

  • maxsize: 缓存的最大条目数，超出这个数量时，会删除最久未使用的条目。默认为128，设为None表示无限制。
  • typed: 如果为True，不同类型的函数参数将被单独缓存（例如，f(3)和f(3.0)会被视为不同的调用）。默认为False。
提示

在Python 3.9及以上版本，还可以使用functools.cache装饰器，它是lru_cache(maxsize=None)的简化版本，用于无限制的缓存。

缓存信息和缓存控制​

lru_cache装饰的函数会获得一些额外的属性和方法，用于查看和管理缓存：

# 继续上面的例子
print("\n缓存信息:")
print(f"缓存命中次数: {fibonacci.cache_info().hits}")
print(f"缓存未命中次数: {fibonacci.cache_info().misses}")
print(f"缓存最大容量: {fibonacci.cache_info().maxsize}")
print(f"当前缓存大小: {fibonacci.cache_info().currsize}")

# 清除缓存
fibonacci.cache_clear()
print("\n清除缓存后:")
print(f"缓存信息: {fibonacci.cache_info()}")

输出结果大致为：

缓存信息:
缓存命中次数: 68
缓存未命中次数: 36
缓存最大容量: 128
当前缓存大小: 36

清除缓存后:
缓存信息: CacheInfo(hits=0, misses=0, maxsize=128, currsize=0)

自定义缓存实现​

虽然lru_cache适用于大多数情况，但有时我们可能需要自定义缓存行为。下面是一个简单的手动实现缓存的例子：

def memoize(func):
    """一个简单的缓存装饰器"""
    cache = {}
    
    def wrapper(*args, **kwargs):
        # 创建键值，将args和kwargs转换为可哈希的类型
        key = str(args) + str(sorted(kwargs.items()))
        if key not in cache:
            cache[key] = func(*args, **kwargs)
        return cache[key]
    
    wrapper.cache = cache  # 为了能够访问和管理缓存
    return wrapper

# 使用自定义缓存装饰器
@memoize
def factorial(n):
    """计算阶乘"""
    if n == 0 or n == 1:
        return 1
    return n * factorial(n-1)

# 测试自定义缓存
print("\n计算阶乘:")
print(f"5! = {factorial(5)}")
print(f"10! = {factorial(10)}")
print(f"缓存内容: {factorial.cache}")

输出结果：

计算阶乘:
5! = 120
10! = 3628800
缓存内容: {'(0,){}': 1, '(1,){}': 1, '(2,){}': 2, '(3,){}': 6, '(4,){}': 24, '(5,){}': 120, '(6,){}': 720, '(7,){}': 5040, '(8,){}': 40320, '(9,){}': 362880, '(10,){}': 3628800}

何时使用函数缓存​

函数缓存非常有用，但并非所有情况都适合使用：

适合使用缓存的情况​

  1. 纯函数：对于相同的输入总是产生相同输出的函数
  2. 计算密集型函数：执行需要大量计算资源的函数
  3. 频繁调用相同参数：当函数使用相同的参数多次调用时
  4. 递归函数：可以显著减少递归调用的次数

不适合使用缓存的情况​

  1. 依赖外部状态的函数：结果受外部因素影响的函数
  2. 输入参数变化大：函数总是使用不同的参数调用
  3. 内存敏感应用：缓存可能会占用大量内存
  4. 副作用函数：有副作用（如修改文件、数据库等）的函数

实际应用案例​

案例1：Web API响应缓存​

import requests
import functools
import time

@functools.lru_cache(maxsize=32)
def fetch_api_data(url):
    """从API获取数据，并缓存结果"""
    print(f"正在从API获取数据: {url}")
    response = requests.get(url)
    return response.json()

# 模拟API请求
def get_user_data(user_id):
    url = f"https://jsonplaceholder.typicode.com/users/{user_id}"
    return fetch_api_data(url)

# 第一次调用 - 将会真正发送请求
print("首次请求用户1的数据:")
start = time.time()
user1 = get_user_data(1)
print(f"请求耗时: {time.time() - start:.4f}秒")
print(f"用户名: {user1['name']}")

# 第二次调用 - 将使用缓存
print("\n再次请求用户1的数据:")
start = time.time()
user1_again = get_user_data(1)
print(f"请求耗时: {time.time() - start:.4f}秒")
print(f"用户名: {user1_again['name']}")

# 请求不同的用户数据
print("\n请求用户2的数据:")
start = time.time()
user2 = get_user_data(2)
print(f"请求耗时: {time.time() - start:.4f}秒")
print(f"用户名: {user2['name']}")

# 查看缓存信息
print(f"\n缓存信息: {fetch_api_data.cache_info()}")

案例2：图像处理​

import functools
from PIL import Image, ImageFilter
import time

@functools.lru_cache(maxsize=20)
def apply_filter(image_path, filter_type, intensity):
    """对图像应用滤镜并缓存结果"""
    print(f"处理图像: {image_path}, 滤镜: {filter_type}, 强度: {intensity}")
    
    # 这里只是演示，实际代码中会读取并处理图像
    # image = Image.open(image_path)
    # 根据filter_type和intensity应用不同的滤镜
    # processed_image = image.filter(...)
    
    # 模拟处理时间
    time.sleep(2)
    return f"已处理的图像: {image_path} (滤镜: {filter_type}, 强度: {intensity})"

# 测试图像处理
print("首次处理图片:")
start = time.time()
result1 = apply_filter("photo.jpg", "BLUR", 5)
print(f"处理时间: {time.time() - start:.2f}秒")

print("\n再次处理相同的图片和参数:")
start = time.time()
result2 = apply_filter("photo.jpg", "BLUR", 5)
print(f"处理时间: {time.time() - start:.2f}秒")

print("\n处理相同的图片但使用不同参数:")
start = time.time()
result3 = apply_filter("photo.jpg", "SHARPEN", 3)
print(f"处理时间: {time.time() - start:.2f}秒")

print(f"\n缓存信息: {apply_filter.cache_info()}")

高级缓存技巧​

设置过期的缓存​

有时我们希望缓存能在一段时间后自动过期。以下是一个带过期时间的缓存装饰器示例：

import functools
import time
from datetime import datetime, timedelta

def timed_lru_cache(seconds=10, maxsize=128):
    """带过期时间的LRU缓存装饰器"""
    def wrapper_cache(func):
        @functools.lru_cache(maxsize=maxsize)
        def wrapped_func(args_key, kwargs_key):
            return func(*args_key[0], **kwargs_key[0])

        # 添加过期时间
        cached_func = wrapped_func
        cached_func.lifetime = seconds
        cached_func.expiration = datetime.now() + timedelta(seconds=seconds)
        
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 检查是否过期
            if datetime.now() >= cached_func.expiration:
                cached_func.cache_clear()
                cached_func.expiration = datetime.now() + timedelta(seconds=cached_func.lifetime)
            
            # 将参数转换为可哈希类型
            args_key = (args,)
            kwargs_key = (frozenset(kwargs.items()),)
            
            return cached_func(args_key, kwargs_key)
        
        # 保留原始函数的一些属性
        wrapper.cache_info = cached_func.cache_info
        wrapper.cache_clear = cached_func.cache_clear
        return wrapper
    
    return wrapper_cache

# 使用带过期时间的缓存
@timed_lru_cache(seconds=5)
def get_current_time():
    """获取当前时间，结果将缓存5秒"""
    return datetime.now().strftime("%H:%M:%S")

# 测试带过期时间的缓存
print("测试带过期时间的缓存:")
for i in range(3):
    print(f"调用 {i+1}: {get_current_time()}")
    time.sleep(2)

print("\n等待缓存过期...")
time.sleep(3)  # 总共已过去9秒，超过了5秒缓存时间

print(f"调用 4: {get_current_time()}")  # 缓存应该已经过期

总结​

函数缓存是Python中一种强大的优化技术，它可以：

  1. 通过存储函数调用结果，避免重复计算来提高性能
  2. 使用@functools.lru_cache轻松实现缓存功能
  3. 特别适用于递归函数、计算密集型函数和重复调用的函数
  4. 可以根据需求自定义缓存行为，如添加过期时间

使用函数缓存时需要权衡内存使用和性能提升，确保在适当的场景中应用这一技术。对于初学者来说，lru_cache是一个非常有用的工具，可以帮助你编写更高效的Python代码。

练习题​

  1. 使用lru_cache装饰器优化一个计算第n个质数的函数。
  2. 实现一个带有过期时间的缓存装饰器，用于缓存从API获取的数据。
  3. 比较带缓存和不带缓存的递归实现的性能差异，例如计算组合数。
  4. 设计一个可以显示缓存命中率的缓存装饰器。
  5. 使用缓存优化一个图像处理函数，处理不同参数下的图像效果。

附加资源​

提示

缓存是一种权衡技术 - 它用空间（内存）换取时间（计算速度）。在资源受限的环境中使用缓存时，务必考虑内存限制。