Python协程中使用lru_cache报错的原因及实现方法

错误原因 #

functools.lru_cache 是为同步函数设计的装饰器，它直接缓存被装饰异步函数返回的协程对象，而非协程执行后的结果：

• 首次调用时，你await该协程对象获取到结果；
• 第二次从缓存取出的是已经被await过的协程对象，而协程对象是一次性的，无法重复执行，因此触发 RuntimeError: cannot reuse already awaited coroutine。

解决方法 #

方法1：手动实现结果缓存 #

自己维护缓存字典，直接缓存协程执行后的结果，同时控制缓存大小：

def __init__(self):
    self._company_id_cache = {}

async def get_company_id(self, simulation_id: int):
    if simulation_id in self._company_id_cache:
        return self._company_id_cache[simulation_id]
    
    simulation_in_db = await self.get_by_id(_id=simulation_id)
    if not simulation_in_db:
        raise ValueError("Simulation not found")
    
    company_id = simulation_in_db["company_id"]
    self._company_id_cache[simulation_id] = company_id
    
    # 控制缓存最大数量为8，超过则删除最早的条目
    if len(self._company_id_cache) > 8:
        oldest_key = next(iter(self._company_id_cache.keys()))
        del self._company_id_cache[oldest_key]
    
    return company_id

方法2：使用专门的异步缓存库async-lru #

async-lru是专为异步函数设计的缓存装饰器，用法和lru_cache类似：

1. 先安装库：

pip install async-lru

2. 修改代码：

from async_lru import alru_cache

@alru_cache(maxsize=8)
async def get_company_id(self, simulation_id: int):
    simulation_in_db = await self.get_by_id(_id=simulation_id)
    if not simulation_in_db:
        raise ValueError("Simulation not found")
    company_id = simulation_in_db["company_id"]
    return company_id

方法3：自定义异步缓存装饰器（不依赖第三方库） #

手动实现一个装饰器，缓存协程的执行结果而非协程对象：

import functools
import asyncio

def async_lru_cache(maxsize=128):
    def decorator(func):
        cache = functools.lru_cache(maxsize=maxsize)
        
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            key = (args, frozenset(kwargs.items()))
            future = cache.get(key)
            
            if future is None:
                future = asyncio.create_task(func(*args, **kwargs))
                cache(*args, **kwargs)
            
            try:
                return await future
            except Exception:
                # 任务执行失败时清除缓存，下次重新尝试
                cache.cache_clear()
                raise
        
        return wrapper
    return decorator

# 使用自定义装饰器
@async_lru_cache(maxsize=8)
async def get_company_id(self, simulation_id: int):
    simulation_in_db = await self.get_by_id(_id=simulation_id)
    if not simulation_in_db:
        raise ValueError("Simulation not found")
    company_id = simulation_in_db["company_id"]
    return company_id

内容的提问来源于stack exchange，提问作者Diego L