FastAPI+Beanie异步环境下并行执行耗时函数的解决方案

核心问题分析 #

你遇到的「不同事件循环」错误，本质是异步操作（如Beanie的ODM调用）绑定了原事件循环，而你尝试在新循环/线程里执行这些操作，导致上下文冲突。同时之前的方案存在用法错误：

• asyncio.to_thread传了协程对象而非函数本身；
• asyncio.run在已有事件循环的FastAPI环境中调用，会强制创建新循环引发冲突；
• 手动创建线程+新循环的方式，破坏了Beanie与原循环绑定的MongoDB连接上下文。

正确实现方案 #

根据需求分两种场景：

场景1：并行执行，等待结果后返回响应（阻塞当前请求，但两个函数彼此并行） #

如果需要确保两个任务完成后再给客户端返回结果，直接用asyncio.gather并行调度异步函数：

from fastapi import FastAPI
import asyncio
from beanie import Document

# 假设你的路由和函数定义
app = FastAPI()

async def add_attack_shodan_results_to_scans_collection(cve_list):
    # 你的异步逻辑（比如Beanie数据库操作）
    ...

async def make_safe_dict_for_mongo_insertion_for_nmap(ip, scan_id):
    # 你的异步逻辑（比如Beanie数据库操作）
    ...

@app.post("/run-scans")
async def run_scans(body: YourRequestModel):
    inserted_id = ... # 假设已获取插入ID
    cve_list_with_scan_id = ... # 假设已准备好参数

    # 并行执行两个异步函数，等待两者完成
    await asyncio.gather(
        add_attack_shodan_results_to_scans_collection(cve_list_with_scan_id),
        make_safe_dict_for_mongo_insertion_for_nmap(body.ip, inserted_id)
    )

    return {"status": "all tasks completed"}

场景2：后台运行，不阻塞当前请求（立即返回响应，任务在后台执行） #

如果不需要等待任务完成，想让客户端立刻收到响应，用asyncio.create_task将任务丢到当前事件循环后台：

@app.post("/run-scans-background")
async def run_scans_background(body: YourRequestModel):
    inserted_id = ...
    cve_list_with_scan_id = ...

    # 提交后台任务，无需await，事件循环会自动调度执行
    asyncio.create_task(
        add_attack_shodan_results_to_scans_collection(cve_list_with_scan_id)
    )
    asyncio.create_task(
        make_safe_dict_for_mongo_insertion_for_nmap(body.ip, inserted_id)
    )

    return {"status": "background tasks started"}

特殊情况：函数包含同步耗时操作（CPU密集/阻塞IO） #

如果你的函数里有同步的耗时代码（比如复杂计算、阻塞式API调用），用asyncio.to_thread把同步部分包装，避免阻塞事件循环：

async def add_attack_shodan_results_to_scans_collection(cve_list):
    # 把同步耗时逻辑封装成单独函数
    def sync_process_cve(cve_data):
        # 这里是CPU密集或阻塞的同步代码
        processed_data = ...
        return processed_data

    # 用to_thread在单独线程执行同步逻辑，不阻塞事件循环
    processed_result = await asyncio.to_thread(sync_process_cve, cve_list)

    # 后续的Beanie异步数据库操作仍在原事件循环执行
    await YourScanCollection.insert_one(processed_result)

关键注意事项 #

1. 所有依赖Beanie/MongoDB的异步操作，必须在FastAPI启动的同一个事件循环中执行，禁止手动创建新循环或用asyncio.run；
2. asyncio.gather用于需要等待结果的并行场景，asyncio.create_task用于后台异步任务；
3. 仅当存在同步耗时操作时才需要用asyncio.to_thread，纯异步IO操作直接用事件循环调度即可。

内容的提问来源于stack exchange，提问作者lEgeL ÇOdēR