FastAPI接口执行多PynamoDB查询触发BrokenResourceError问题求助

问题现象 #

单个API请求中执行多轮DynamoDB（PynamoDB）查询时，单轮查询无异常，但多轮查询会触发BrokenResourceError。虽能获取预期返回值，但该错误会导致后端挂起，阻塞其他请求处理。

已排查的触发原因 #

根据相关文档，BrokenResourceError的触发场景包括：

• 资源被其他任务关闭或释放
• 资源被外部进程或系统关闭或释放
• 资源使用过程中出错，无法继续运行

结合自身场景判断，属于第一种情况，可能的触发场景：

• 多任务间共享DynamoDB资源实例，某任务使用完毕后关闭了资源
• 使用资源的任务被取消或中断，导致关联资源被关闭
• 资源所属任务组被取消，所有关联资源被批量关闭

已尝试的解决方案 #

• 替换为Boto3直接调用DynamoDB，问题依旧存在
• 将查询拆分到多个独立API请求中，此方法可行但不符合业务需求
• 尝试一次查询获取所有目标数据，仍触发相同错误，怀疑是请求总耗时过长导致

触发异常的示例代码 #

@my_router.get("/my_endpoint")
async def my_endpoint(
):
    attributes = []
    for x in range(0, 10):
        attributes.append(PynamoDBClass.get(
            hash_key="hash_key",
            range_key="range_key",
            consistent_read=True,
        ).attribute_values)
    return attributes

可行解决方向 #

1. 避免共享客户端实例：异步环境下DynamoDB客户端（PynamoDB/Boto3）并非线程安全，多任务共享实例易导致资源被意外关闭。建议为每个请求或任务创建独立的客户端实例。
2. 改用异步并发执行查询：当前同步循环执行查询的方式会拉长总耗时，改用asyncio.gather并发执行所有查询，减少整体请求时长，避免因超时触发资源回收。示例代码：

import asyncio

@my_router.get("/my_endpoint")
async def my_endpoint(
):
    async def fetch_item():
        item = await PynamoDBClass.get(
            hash_key="hash_key",
            range_key="range_key",
            consistent_read=True,
        )
        return item.attribute_values
    
    tasks = [fetch_item() for _ in range(10)]
    attributes = await asyncio.gather(*tasks)
    return attributes

3. 调整客户端超时配置：检查DynamoDB客户端的超时参数，适当延长超时阈值，避免因单轮查询耗时过长触发资源回收。
4. 捕获异常并隔离影响：在查询逻辑中捕获BrokenResourceError，及时重新创建客户端实例或释放资源，避免错误扩散导致整个后端挂起。

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