FastAPI 后台任务使用指南
FastAPI BackgroundTasks 让耗时操作在响应返回后继续执行：发邮件、处理文件、生成报表。本文讲解基本用法、复杂数据处理、文件上传场景与生产监控策略。
直接回答：FastAPI 的 BackgroundTasks 让你把耗时操作丢到响应返回之后执行——用户立即拿到结果，任务在同一 Web 进程内继续执行，但不保证最终完成。它适合轻量级后台工作；重型任务请交给 Celery 这类任务队列。
解决什么问题
同步处理的时代，用户点个"导出报表"要等 30 秒转圈。后台任务把模式翻转过来：接口立即返回"已受理"，慢活在后台慢慢干。可缩短响应等待，但仍占用服务资源，不保证吞吐提升。
第一个后台任务
from fastapi import FastAPI, BackgroundTasks, UploadFile
app = FastAPI()
def send_welcome_email(email: str):
# 模拟耗时发送
print(f"发送欢迎邮件给 {email}")
@app.post("/signup")
def signup(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(send_welcome_email, email)
return {"msg": "注册成功，欢迎邮件稍后送达"}
BackgroundTasks 由 FastAPI 依赖注入提供，add_task 登记的任务会在响应发出后依次执行。多个任务按登记顺序运行。
传递复杂数据
任务函数就是普通 Python 函数，参数随意：
def generate_report(user_id: int, period: str, recipients: list[str]):
report = build_report(user_id, period)
for r in recipients:
deliver(report, r)
@app.post("/reports")
def create_report(req: ReportRequest, background_tasks: BackgroundTasks):
background_tasks.add_task(generate_report, req.user_id, req.period, req.recipients)
return {"msg": "报表生成中", "status": "processing"}
实践建议：任务参数传"数据 ID"而非大对象本体，任务函数自己去数据库取——避免大对象及已关闭的请求数据库会话；BackgroundTasks 本身不序列化参数。
文件上传场景
上传大文件后需要做病毒扫描、转码、生成缩略图时：
@app.post("/upload")
async def upload(file: UploadFile, background_tasks: BackgroundTasks):
path = await save_temp(file)
background_tasks.add_task(scan_and_process, path)
return {"filename": file.filename, "status": "已受理，处理中"}
此处仍需等待 save_temp 完成上传保存，不能承诺秒响应；save_temp、scan_and_process 是项目实现的占位函数。后台应接收持久化路径或对象键，不应依赖请求结束后可能已关闭的 UploadFile；另需大小限制、清理和失败状态。处理结果可通过另行实现的轮询或 WebSocket 通知。
边界：什么时候该上任务队列
| 场景 | 选择 |
|---|---|
| 发邮件、写日志、清理缓存 | BackgroundTasks 足够 |
| 必须保证执行（失败重试） | Celery / ARQ / Dramatiq |
| 跨进程、跨机器分发 | 任务队列 |
| 需要定时调度 | APScheduler 或队列的 beat |
| 任务执行时间超过请求超时很多 | 任务队列 |
BackgroundTasks 与 Web 进程同生死：进程重启，未执行的任务就丢了。持久化队列仍需确认、重试、死信和幂等设计，不自动保证恰好执行一次。
生产监控策略
后台任务的失败不会体现在 HTTP 状态码里，必须自己埋点：任务开始/结束/失败各记一条结构化日志，关键任务记录执行耗时。同时保存待执行任务的业务状态，才能区分成功、失败和未完成。集中排查时，以任务 ID 串起开始与结束日志；观测云的日志检测可针对已采集到的失败日志设置规则，未完成任务则应结合业务状态核对。
常见问题（FAQ）
Q：async def 和 def 的任务函数有区别吗？
A：有。def 任务在线程池执行（不阻塞事件循环），async def 任务在事件循环里 await。短小同步阻塞任务可用 def，纯异步 IO 用 async def；重型 CPU 任务应交给独立进程或 worker。
Q：BackgroundTasks 里抛异常会怎样？
A：异常不会传给客户端（响应已发出），异常会向服务端传播，同一 BackgroundTasks 中后续任务将不再执行。任务函数内部务必 try/except 并记录日志，否则失败无声无息。
Q：能给任务加进度查询吗？
A：BackgroundTasks 本身没有进度机制。常见做法：任务函数把进度写入 Redis/数据库，前端轮询进度接口；需要实时推送就用 WebSocket。
官方参考
本文基于官方文档整理，未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全；局部片段不等同于完整生产应用。
