Python functools.lru_cache的缓存淘汰策略与线程安全

Python 标准库中的 functools.lru_cache 是一个将函数结果进行缓存的装饰器。通过缓存，它能用“空间换时间”的策略，显著减少重复计算或 I/O 操作的开销。本文将直接讲解如何配置其淘汰策略，以及它在多线程环境下的安全特性。

理解 LRU 缓存淘汰机制

LRU (Least Recently Used) 算法的核心逻辑是：当缓存空间已满时，优先淘汰“最久未被使用”的数据。其内部通常结合哈希表（字典）和双向链表来实现，以保证查找和更新的效率。

以下是 LRU 缓存处理一次新数据请求的逻辑流程：

步骤 1：基础用法与效果对比

通过一个计算密集型的例子，直观感受缓存带来的性能差异。

1. 打开 Python 编辑器或 IDE。
2. 导入 time 和 functools 模块。
3. 定义 一个未使用缓存的递归函数 fib_no_cache。
4. 定义 一个使用 @lru_cache 装饰器的函数 fib_with_cache。
5. 运行 下面的代码查看耗时对比。

import time
from functools import lru_cache

# 未使用缓存的版本
def fib_no_cache(n):
    if n < 2:
        return n
    return fib_no_cache(n - 1) + fib_no_cache(n - 2)

# 使用缓存的版本
@lru_cache(maxsize=None)
def fib_with_cache(n):
    if n < 2:
        return n
    return fib_with_cache(n - 1) + fib_with_cache(n - 2)

# 测试未缓存版本
start = time.time()
res1 = fib_no_cache(35)
t1 = time.time() - start

# 测试缓存版本
start = time.time()
res2 = fib_with_cache(35)
t2 = time.time() - start

print(f"无缓存耗时: {t1:.6f}秒")
print(f"有缓存耗时: {t2:.6f}秒")

运行结果显示，使用缓存的版本耗时通常在微秒级，而无缓存版本由于大量重复计算，耗时可能达到秒级。

步骤 2：配置缓存淘汰策略

@lru_cache 装饰器接受两个主要参数，用于控制缓存的大小和类型区分。

参数配置表

2.1 限制缓存大小 (maxsize)

当 maxsize 设置为一个正整数（例如 128）时，一旦缓存的条目超过该数值，系统会自动移除最久未使用的条目。

1. 修改 装饰器参数，限制缓存大小为 32。
2. 观察 cache_info() 的输出，查看 currsize 是否超过 32。

@lru_cache(maxsize=32)
def process_data(data_id):
    # 模拟耗时操作
    time.sleep(0.01)
    return data_id * 10

# 循环调用多次
for i in range(100):
    process_data(i)

# 查看缓存状态
print(process_data.cache_info())

2.2 区分参数类型 (typed)

默认情况下，Python 认为 3 (整型) 和 3.0 (浮点型) 是相同的参数值，命中缓存时会直接返回结果。如果你需要严格区分类型：

1. 设置 typed=True。
2. 调用 函数传入不同类型的参数。

@lru_cache(maxsize=128, typed=True)
def check_type(num):
    return f"Type is {type(num).__name__}"

print(check_type(1))    # 输出: Type is int
print(check_type(1.0))  # 输出: Type is float (视为一次新的未命中调用)

步骤 3：监控与维护缓存

在调试或优化性能时，需要查看缓存的命中率或手动清除脏数据。

3.1 查看缓存统计 (cache_info)

该函数返回一个包含命名元组的对象，展示了缓存的运行状态。

返回字段说明

查看统计信息

1. 调用 函数名加 .cache_info()。
2. 分析 hits 与 misses 的比例，比例越高说明缓存效果越好。

print(process_data.cache_info())
# 示例输出: CacheInfo(hits=20, misses=80, maxsize=32, currsize=32)

3.2 清除缓存 (cache_clear)

当数据源发生变化，需要强制函数重新计算时，必须清除缓存。

1. 调用 函数名加 .cache_clear()。
2. 验证 再次调用 cache_info()，确认 currsize 归零。

process_data.cache_clear()
print(process_data.cache_info())
# 输出: CacheInfo(hits=0, misses=0, maxsize=32, currsize=0)

步骤 4：理解线程安全特性

在多线程环境下使用缓存是常见场景，functools.lru_cache 在设计上考虑了线程安全。

4.1 底层锁机制

lru_cache 的内部实现使用了线程锁。这意味着在多线程同时调用同一个被装饰的函数时：

• 不同的参数（并发写入）：内部锁机制确保了缓存的更新操作是原子的，不会出现数据覆盖或链表指针错误。
• 相同的参数（并发读取）：如果一个线程正在计算该参数的值，其他线程不会重复计算，而是等待计算完成后直接获取结果。

4.2 统计数据的精度

虽然缓存操作是线程安全的，但 cache_info() 返回的统计数据（hits 和 misses）在极高并发的场景下是近似值。

• 原因：为了保证性能，计数器的增加可能没有在极其严格的原子锁下进行，避免锁竞争导致性能下降。
• 结论：可以信赖缓存结果的一致性，但不要将其统计数据用于严格的精确计数逻辑。

多线程调用示例

1. 导入 threading 模块。
2. 创建 多个线程同时调用缓存函数。
3. 检查 结果正确性及缓存统计。

from threading import Thread

@lru_cache(maxsize=None)
def heavy_task(n):
    time.sleep(0.01)
    return n * n

def worker(n):
    result = heavy_task(n)
    print(f"Thread processed {n}, got {result}")

threads = []
# 启动10个线程，其中一半计算相同的值
for i in range(10):
    t = Thread(target=worker, args=(i % 5,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(heavy_task.cache_info())

即使在多线程下，相同的 n 值也只会导致一次实际的 heavy_task 执行（misses 较少），其余均为缓存命中。

步骤 5：注意事项与最佳实践

虽然 lru_cache 很强大，但滥用会导致内存泄漏或逻辑错误。

1. 确保参数可哈希：由于使用字典作为存储，函数的参数必须是不可变类型（如整数、字符串、元组）。禁止传入列表、字典等可变对象。
2. 避免用于副作用函数：如果函数不仅仅返回值，还进行了文件写入、发送邮件等副作用操作，缓存会导致这些操作在命中时不执行，引发逻辑错误。
3. 关注内存增长：如果设置了 maxsize=None，且参数空间无限（例如递归深度极大或随机参数），内存可能被占满。建议 生产环境始终设置合理的 maxsize。
4. 获取原始函数：如果需要绕过缓存直接调用原始函数，访问 .__wrapped__ 属性。

# 绕过缓存直接调用
raw_result = fib_with_cache.__wrapped__(10)