FastAPI中asyncio.gather在后台任务中异常挂起的问题

本地使用命令 uvicorn server.app:fapi --reload --host 0.0.0.0 --port 80 运行FastAPI应用，想要同时运行3个协程，但遇到asyncio.gather()和await asyncio.sleep(3)挂起的情况：执行代码后只有write_notification的初始打印，sleep后的输出和process_notifications里的打印完全没出现，就算移除await asyncio.sleep(3)调试也没用。

相关代码 #

import asyncio
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

async def write_notification(email: str, message=""):
    print(f'Processing message "{message}"')
    await asyncio.sleep(3) # 模拟发送邮件
    print('foo after sleep') # <-- 从未打印
    return message

async def process_notifications(email: str, message=""):
    tasks = [write_notification(email, f'{message} - {i}') for i in range(3)] # 3个任务入队
    result = await asyncio.gather(*tasks)
    print(result) # <-- 从未打印
    for i in result:
        print(i, 'sent') # <-- 从未打印

@app.get('/bg-tasks-multiple', status_code=201)
async def background_tasks(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(process_notifications, email, f"notification for {email}")
    return {'success': True, 'email': email}

执行输出 #

Processing message "notification for fef@ewf.fgwe - 0"
Processing message "notification for fef@ewf.fgwe - 1"
Processing message "notification for fef@ewf.fgwe - 2"

问题原因与解决方法 #

问题核心：FastAPI的BackgroundTasks不支持直接添加异步函数作为后台任务。当你调用add_task传入异步函数时，它只会执行到第一个await就暂停，不会帮你启动事件循环处理后续的异步逻辑，导致协程永远无法完成。

解决方法一：用asyncio.run()手动触发异步任务执行

修改接口中的后台任务添加逻辑，用asyncio.run()包裹异步函数调用：

@app.get('/bg-tasks-multiple', status_code=201)
async def background_tasks(email: str, background_tasks: BackgroundTasks):
    # 用asyncio.run()启动事件循环，完整执行异步任务
    background_tasks.add_task(asyncio.run, process_notifications(email, f"notification for {email}"))
    return {'success': True, 'email': email}

解决方法二：把异步函数包装成同步函数

先写一个同步函数，内部用asyncio.run()执行异步逻辑，再把这个同步函数添加到后台任务：

def sync_process_notifications(email: str, message=""):
    asyncio.run(process_notifications(email, message))

@app.get('/bg-tasks-multiple', status_code=201)
async def background_tasks(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(sync_process_notifications, email, f"notification for {email}")
    return {'success': True, 'email': email}

注意事项 #

FastAPI的BackgroundTasks是轻量级实现，适合短耗时的后台任务。如果是需要长时间运行、重试或分布式处理的任务，建议用Celery、RQ这类专业的任务队列框架。

内容的提问来源于stack exchange，提问作者Manuel Simões