FastAPI的BackgroundTasks不捕获异常，会导致任务静默失败；必须手动为每个任务函数添加try/except并记录结构化日志；它不支持重试，关键任务应改用Celery等专业队列。
BackgroundTasks 抛异常会导致任务丢失
FastAPI 的 BackgroundTasks 本质是把函数加到事件循环末尾异步执行，**不捕获异常**。一旦任务内部抛出未处理异常，它就静默失败，既不会重试，也不会通知主请求流程，更不会回滚关联操作。
典型现象：你在 background_tasks.add_task(send_email, user_id) 后返回 HTTP 200，但邮件根本没发出去，日志里也看不到错误——因为异常被吞掉了。
- 异常发生在后台任务线程/协程中，与主请求生命周期解耦
- FastAPI 不提供内置重试、死信队列或失败回调机制
-
add_task()调用本身成功，不代表任务逻辑成功
必须手动包装任务函数并记录异常
最直接可靠的做法是：**所有传给 add_task() 的函数，都得自己包一层 try/except**，至少确保异常可观察。不要依赖外部监控被动发现失败。
示例：
def safe_send_email(user_id: int):
try:
send_email_impl(user_id) # 真正的业务逻辑
except Exception as e:
# 记入结构化日志（别只 print！）
logger.error("Background task failed", extra={"task": "send_email", "user_id": user_id, "error": str(e)})
# 可选：发告警、写 DB 失败记录、调用 webhook
mark_email_failed_in_db(user_id)
@app.post("/users")
async def create_user(background_tasks: BackgroundTasks):
user = await save_user_to_db()
background_tasks.add_task(safe_send_email, user.id)
return {"id": user.id}
- 不要在
send_email_impl内部 try/catch——它可能被复用在非 background 场景，职责要分离 - 日志必须带上下文字段（如
user_id），否则排查时无法关联请求和任务 - 避免在 except 块里再抛异常——这会中断 event loop，影响后续 background 任务
需要重试？得自己实现或换方案
FastAPI 原生 BackgroundTasks **不支持重试**。如果任务失败后必须重试（比如第三方 API 临时超时），有两条路：
- 在
safe_send_email里手动加指数退避重试逻辑（适合简单、低频、无状态任务） - 改用专业任务队列：Celery + Redis/RabbitMQ，或 Dramatiq，它们原生支持重试、延迟、优先级、监控面板
例如 Celery 方式：
@celery.task(bind=True, autoretry_for=(Exception,), retry_kwargs={'max_retries': 3, 'countdown': 60})
def send_email_task(self, user_id: int):
send_email_impl(user_id)
注意：BackgroundTasks 和 Celery 是互斥方案，不是叠加关系。一旦选了 Celery，就不再用 BackgroundTasks.add_task()。
更隐蔽的坑：异步任务里的同步阻塞调用
如果你的任务函数是 async def，但里面用了 time.sleep()、requests.get() 这类同步阻塞调用，会导致整个 event loop 卡住——不仅当前任务失败，还可能拖垮其他 background 任务甚至主接口。
- 用
asyncio.sleep()替代time.sleep() - 用
httpx.AsyncClient或aiohttp替代requests - 不确定是否异步的库？查文档，或加
asyncio.to_thread()包装（Python 3.9+）
这类问题不会抛异常，但会让任务“假死”，比直接报错更难定位。
真正关键的不是“怎么让 BackgroundTasks 支持异常重试”，而是清醒意识到它只是个简易的 fire-and-forget 工具——重要、需保障、有状态、要追溯的任务，从来就不该放在这里跑。
