该教程介绍如何设计一个自定义装饰器 @iterator_cache,使其对函数输入的每个元素单独缓存(而非整个元组),从而避免重复调用高开销函数,显著提升批量处理性能。
该教程介绍如何设计一个自定义装饰器 `@iterator_cache`,使其对函数输入的**每个元素单独缓存**(而非整个元组),从而避免重复调用高开销函数,显著提升批量处理性能。
在实际开发中,我们常遇到这样一类场景:某个外部库提供的函数(如 expensive_eval)对单个输入开销巨大(含固定延迟 + 可变计算),而业务层需频繁调用它处理可迭代对象(如 list[int] 或 tuple[int])。若直接对整个输入元组缓存(如用 @lru_cache),则 [1,2] 和 [1,2,3] 会被视为完全不同的键,无法复用已计算的 1 和 2 的结果;但若逐元素缓存,则能真正实现“一次计算、多次复用”。
核心思路是:将输入序列拆解为“已缓存元素”和“未缓存元素”,仅对后者调用昂贵函数,再将新结果合并进全局缓存,最终按原顺序组装返回值。
以下是完整的 @iterator_cache 装饰器实现:
from collections.abc import Sequence, Iterable
from functools import wraps
from typing import Any, Callable, Dict, List, TypeVar, Union
T = TypeVar('T')
R = TypeVar('R')
def iterator_cache(
expensive_func: Callable[[Sequence[T]], Sequence[R]]
) -> Callable[[Iterable[T]], List[R]]:
"""
创建一个装饰器,使目标函数对输入中的每个元素独立缓存结果。
Args:
expensive_func: 原始高开销函数,接受序列并返回对应序列结果
Returns:
装饰后的函数:接受任意可迭代对象,返回列表形式结果
"""
cache: Dict[T, R] = {}
def decorator(func: Callable[[Iterable[T]], List[R]]) -> Callable[[Iterable[T]], List[R]]:
@wraps(func)
def wrapper(iterable: Iterable[T]) -> List[R]:
# 转为 list 以支持多次遍历(如用于去重和结果组装)
items = list(iterable)
if not items:
return []
# 分离出未缓存的元素
uncached = [x for x in items if x not in cache]
# 仅当存在新元素时才调用昂贵函数(避免空输入触发无意义延迟)
if uncached:
# 注意:expensive_func 必须接受 Sequence(如 list/tuple),故传入 uncached(list)
new_results = expensive_func(uncached)
# 将新结果写入缓存
for x, res in zip(uncached, new_results):
cache[x] = res
# 按原始顺序从缓存中取值(保证输出顺序与输入一致)
return [cache[x] for x in items]
return wrapper
return decorator使用方式如下(与问题中示例完全兼容):
import time
def expensive_eval(numbers: tuple[int]) -> tuple[int]:
time.sleep(5 + len(numbers)) # 模拟固定+可变延迟
print(f"expensive eval of {numbers}")
return tuple(n ** 2 for n in numbers)
# 应用装饰器:指定 expensive_eval 作为底层计算函数
@iterator_cache(expensive_eval)
def cheap_eval(numbers: list[int]) -> list[int]:
print(f"cheap eval of {numbers}")
return [] # 实际逻辑由装饰器接管,此处仅作占位
# 测试
print(cheap_eval([1, 2])) # 首次:触发 expensive_eval((1,2))
print(cheap_eval([1, 3])) # 复用 1,仅计算 3 → expensive_eval((3,))
print(cheap_eval([1, 2, 3])) # 全部命中缓存✅ 关键优势:
- 总耗时从 5+2 + 5+2 + 5+3 = 22s 降至 5+2 + 5+1 + 0 = 13s(假设 expensive_eval 延迟为 5 + len(input))
- 缓存粒度精确到元素级,不依赖输入顺序或结构(如 [1,2] 与 (2,1) 中的 1、2 仍可共享结果)
- 线程安全需额外加锁(本例为简化未包含;生产环境建议用 threading.Lock 或 concurrent.futures.ThreadPoolExecutor 控制并发访问)
⚠️ 注意事项:
- 被装饰函数的参数必须是可迭代且元素可哈希(如 int, str, tuple),否则无法作为字典键;
- expensive_func 输入类型需与 uncached 类型一致(通常为 list 或 tuple),注意其签名兼容性;
- 全局缓存 cache 是闭包变量,每个 @iterator_cache(...) 实例拥有独立缓存空间,互不干扰;
- 若需持久化或进程间共享,应替换为 functools.lru_cache、redis 或 sqlite 等方案。
通过该装饰器,你能在不修改第三方库函数的前提下,优雅地实现细粒度缓存优化——让“廉价接口”真正发挥杠杆效应。

















