asyncio 中不能直接用 @lru_cache，因其为同步装饰器，不支持 await 且导致协程未被调用、缓存失效或内存泄漏；应使用 async_lru 库的 @alru_cache 实现协程安全缓存。
asyncio 里不能直接用 @lru_cache
Python 标准库的 @lru_cache 是同步装饰器，它内部用的是普通字典和线程锁，不支持 await 表达式。如果你给一个 async def 函数加 @lru_cache，调用时会直接返回一个 coroutine 对象，而不是执行缓存逻辑——更糟的是，多次调用会生成多个未 await 的协程，导致内存泄漏或意料之外的并发行为。
常见错误现象：
– 缓存没生效，每次调用都走实际逻辑
– 报错 RuntimeWarning: coroutine 'xxx' was never awaited
– 缓存键被当成 coroutine 对象存储，导致 key 冲突或无法命中
- 根本原因：缓存逻辑本身必须在协程上下文中运行，且读写共享状态需协程安全
- 不能靠把
@lru_cache套在 async 函数外“假装有用” - 标准库没有提供
@alru_cache，得自己搭或选可靠第三方
用 async_lru 库最省心
async_lru 是目前最轻量、维护活跃、API 最接近标准 @lru_cache 的方案。它底层用 asyncio.Lock 保证并发安全，支持 maxsize、typed 等参数，还能正确处理位置/关键字参数组合。
安装与基本用法：
立即学习“Python免费学习笔记（深入）”；
pip install async-lru
示例：
from async_lru import alru_cache
<p>@alru_cache(maxsize=128)
async def fetch_user(user_id: int) -> dict:</p><h1>模拟异步 IO</h1><pre class='brush:python;toolbar:false;'>await asyncio.sleep(0.1)
return {"id": user_id, "name": f"user_{user_id}"}正确调用
result = await fetch_user(123)
- 注意：装饰器名是
@alru_cache，不是@async_lru_cache或其他变体 -
maxsize=None表示无限制，但生产环境建议设合理值，避免内存膨胀 - 不支持
cache_clear()的异步版本（原生alru_cache提供cache_clear同步方法，但清空操作本身是线程安全的，无需 await）
手动实现要注意协程锁和键归一化
如果不想引入新依赖，可以基于 functools.lru_cache + asyncio.Lock 手写，但关键点容易漏：
- 缓存存储必须用普通字典（不能存协程对象），所以得把 await 逻辑封装进缓存值的“获取+填充”流程
- 多个协程同时请求同一个 key 时，要确保只触发一次真实调用（即“缓存穿透防护”），否则失去意义
- 键生成必须稳定：
functools._make_key不支持协程参数，需手动对*args和**kwargs做不可变哈希处理（例如tuple(args) + tuple(sorted(kwargs.items()))） - 锁必须是
asyncio.Lock()，不能用threading.Lock，后者会阻塞事件循环
简化的手动结构示意（不推荐直接用于生产）：
import asyncio
from functools import _make_key
<p>_cache = {}
_lock = asyncio.Lock()</p><p>async def my_alru_cache(func):
async def wrapper(*args, **kwargs):
key = _make_key(args, kwargs, typed=False)
async with _lock:
if key in _cache:
return _cache[key]</p><h1>防穿透：锁内检查，锁外执行</h1><pre class='brush:python;toolbar:false;'> result = await func(*args, **kwargs)
async with _lock:
_cache[key] = result
return result
return wrapper</pre>这版缺少过期、大小限制、类型区分等，仅说明核心难点在哪。
缓存失效和调试要盯住 __wrapped__ 和 cache_info()
async_lru 的装饰器对象保留了 __wrapped__ 属性和 cache_info() 方法，但注意：
-
cache_info()返回的是同步统计（hits,misses,maxsize,currsize），它反映的是缓存字典状态，不是协程调度次数 - 调试时别用
inspect.iscoroutinefunction(fetch_user)判断——装饰后它返回False，因为@alru_cache返回的是普通函数（内部 await） - 真正要确认是否命中，加日志比看
cache_info()更可靠：
@alru_cache(maxsize=1)
async def demo(x):
print(f"computing for {x}") # 只有 miss 时才打印
await asyncio.sleep(0.01)
return x * 2异步缓存的复杂点不在“怎么写”，而在“怎么验证它真按预期工作”。尤其当 key 含可变对象（如 dict）、或函数本身有副作用时，缓存行为容易偏离直觉。
