FastAPI的BackgroundTasks异常默认被事件循环丢弃,应在任务函数内用try/except捕获并记录日志;不推荐依赖shutdown钩子或全局异常处理器;需自行封装带异常监听的Task类以实现可靠监控。

BackgroundTasks 的异常默认会被吞掉
FastAPI 的 BackgroundTasks 在请求返回后异步执行,但它的底层是通过 asyncio.create_task() 启动协程,而这些任务一旦抛出未捕获异常,会直接被事件循环丢弃——你既收不到报错日志,也无法在请求上下文中感知失败。
这不是 Bug,是 asyncio 的默认行为:未被 await 的 task 异常不会传播,也不会中断主线程。
用 try/except 包裹 background task 函数体
最直接、最可控的方式是在你注册的 background 函数内部做异常兜底:
- 所有可能出错的逻辑必须放在
try块里 -
except中记录日志(推荐用logging.exception()保留 traceback) - 避免在 background 函数里 raise 新异常,否则照样被吞
import logging from fastapi import BackgroundTaskslogger = logging.getLogger(name)
def send_email_async(user_id: int): try:
模拟可能失败的操作
raise ValueError("SMTP connection failed") except Exception: logger.exception("Failed to send email for user %s", user_id)@app.post("/notify") def notify_user(background_tasks: BackgroundTasks): background_tasks.add_task(send_email_async, user_id=123) return {"status": "accepted"}
不建议依赖 on_shutdown 或 global exception handler
有人想用
app.on_event("shutdown")等待所有 background tasks 结束再检查异常,这不可行:
- FastAPI 不提供对 background task 集合的引用,无法遍历或 await 它们
- shutdown 时 event loop 可能已关闭,
await task会报RuntimeError: Event loop is closed - 全局异常钩子(如
sys.excepthook)对 asyncio task 异常无效,除非你手动调用task.exception()
进阶:封装带异常监听的 BackgroundTask 类
如果你项目中大量使用 background task 且需要统一监控,可以轻量封装一个替代方案:
- 用
asyncio.create_task()显式创建 task,并保存引用 - 在 task 完成后主动调用
task.exception()检查是否出错 - 把异常转发到日志或告警通道(注意别阻塞 event loop)
关键点在于:不能只靠 add_task,得自己管理 task 生命周期。FastAPI 原生 BackgroundTasks 就是“fire-and-forget”,真要可靠,就得自己多走一步。









