Python的functools.lru_cache是内置的LRU（最近最少使用）缓存装饰器，能够自动缓存函数的返回值，当相同参数再次调用函数时直接返回缓存结果，避免重复计算。但在实际开发中，我们常常需要更灵活的缓存能力，比如缓存过期、自定义淘汰规则、支持异步函数等，这就需要对lru_cache进行扩展。
functools.lru_cache基础用法
原生lru_cache的使用非常简单，只需要装饰目标函数即可，它支持设置maxsize参数指定最大缓存数量，设置为None时表示缓存无上限。
import functools
import time
# 使用lru_cache装饰函数，最大缓存128个结果
@functools.lru_cache(maxsize=128)
def compute_square(n):
time.sleep(1) # 模拟耗时计算
return n * n
# 第一次调用，会执行计算
print(compute_square(5)) # 等待1秒后输出25
# 第二次调用相同参数，直接返回缓存结果
print(compute_square(5)) # 立即输出25
扩展场景一：添加缓存过期功能
原生lru_cache没有缓存过期机制，缓存结果会一直存在直到被淘汰。我们可以通过包装lru_cache，给每个缓存结果添加时间戳，在获取缓存时判断是否过期。
import functools
import time
def lru_cache_with_ttl(ttl=60, maxsize=128):
"""带过期时间的lru_cache扩展装饰器，ttl单位为秒"""
def decorator(func):
# 创建原生lru_cache装饰器
cached_func = functools.lru_cache(maxsize=maxsize)(func)
# 存储缓存的时间戳，key和cached_func的缓存key对应
cache_time = {}
def wrapper(*args, **kwargs):
# 生成缓存key，lru_cache的key是args和frozenset(kwargs.items())
key = args
if kwargs:
key += tuple(sorted(kwargs.items()))
current_time = time.time()
# 检查缓存是否存在且未过期
if key in cache_time and current_time - cache_time[key] < ttl:
return cached_func(*args, **kwargs)
# 缓存不存在或已过期，重新计算
result = func(*args, **kwargs)
cached_func.cache_clear() # 清除旧缓存，重新缓存
cached_func(*args, **kwargs) # 存入新结果
cache_time[key] = current_time
return result
return wrapper
return decorator
# 使用示例，缓存有效期10秒
@lru_cache_with_ttl(ttl=10)
def get_data(param):
return f"data_{param}_{time.time()}"
print(get_data("test")) # 第一次调用，生成新数据
time.sleep(5)
print(get_data("test")) # 5秒后调用，缓存未过期，返回相同结果
time.sleep(6)
print(get_data("test")) # 总共11秒后调用，缓存过期，生成新数据
扩展场景二：支持异步函数缓存
原生lru_cache不支持异步函数，直接装饰async函数会返回协程对象而不是缓存结果，需要针对异步函数做适配。
import functools
import asyncio
def async_lru_cache(maxsize=128):
"""支持异步函数的lru_cache扩展装饰器"""
def decorator(func):
# 用原生lru_cache缓存协程的结果，注意这里缓存的是已经执行完成的返回值
cache = functools.lru_cache(maxsize=maxsize)
@functools.wraps(func)
async def wrapper(*args, **kwargs):
key = args
if kwargs:
key += tuple(sorted(kwargs.items()))
# 检查缓存是否存在
if key in cache._cache: # 访问内部缓存字典，注意不同Python版本可能有差异
return cache(key)
# 执行异步函数获取结果
result = await func(*args, **kwargs)
# 存入缓存
cache(key, result)
return result
return wrapper
return decorator
# 使用示例
@async_lru_cache(maxsize=64)
async def async_fetch_data(url):
await asyncio.sleep(1) # 模拟异步请求
return f"content_of_{url}"
async def main():
print(await async_fetch_data("https://ipipp.com/test")) # 等待1秒输出
print(await async_fetch_data("https://ipipp.com/test")) # 立即输出缓存结果
asyncio.run(main())
扩展场景三：自定义淘汰策略
原生lru_cache采用最近最少使用策略淘汰缓存，如果我们需要实现FIFO（先进先出）等其他淘汰策略，可以基于lru_cache的缓存管理接口进行扩展。
import functools
from collections import OrderedDict
def fifo_cache(maxsize=128):
"""FIFO淘汰策略的缓存装饰器"""
def decorator(func):
cache = OrderedDict()
@functools.wraps(func)
def wrapper(*args, **kwargs):
key = args
if kwargs:
key += tuple(sorted(kwargs.items()))
# 命中缓存直接返回
if key in cache:
return cache[key]
# 未命中，执行函数
result = func(*args, **kwargs)
# 缓存达到上限，删除最早加入的缓存
if len(cache) >= maxsize:
cache.popitem(last=False)
cache[key] = result
return result
return wrapper
return decorator
# 使用示例
@fifo_cache(maxsize=3)
def add(a, b):
return a + b
print(add(1, 2)) # 缓存{ (1,2):3 }
print(add(2, 3)) # 缓存{ (1,2):3, (2,3):5 }
print(add(3, 4)) # 缓存{ (1,2):3, (2,3):5, (3,4):7 }
print(add(4, 5)) # 缓存满，删除(1,2)，缓存{ (2,3):5, (3,4):7, (4,5):9 }
扩展注意事项
在对functools.lru_cache进行扩展时，需要注意以下几点：
- 被缓存的函数参数必须是可哈希的，否则无法作为缓存的key，这也是原生lru_cache的要求
- 扩展缓存过期功能时，尽量避免频繁调用
cache_clear()方法，否则会清空所有缓存，影响性能 - 异步缓存扩展中，访问lru_cache的内部缓存属性时，需要注意不同Python版本的兼容性差异
- 自定义淘汰策略时，如果需要保留lru_cache的统计功能，可以手动维护
cache_info相关的属性
通过上述扩展方式，我们可以让functools.lru_cache适配更多业务场景，在提升程序性能的同时满足个性化的缓存需求。
Pythonfunctools_lru_cache缓存机制缓存扩展修改时间：2026-07-16 07:51:36