FastAPI 背景任務全解析：異步執(zhí)行長任務的正確姿勢！
你是否遇到這種場景：
- 注冊接口響應太慢，只因要發(fā)送郵件？
- 每次寫日志都阻塞了響應返回？
- 想跑個后臺任務，卻不知從哪開始？
這篇文章，帶你徹底掌握 FastAPI 的后臺任務機制（Background Tasks）：它不是 async，不是 Celery，但輕巧高效，正是你項目中最實用的異步利器之一！
什么是 Background Tasks？
Background Tasks 是 FastAPI 提供的一種輕量異步任務執(zhí)行方式。
它的核心思路：
- 在 HTTP 請求響應返回之后，自動執(zhí)行一段你注冊好的函數(shù)
- 適用于不需要立即返回結果、也不強依賴成功率的任務
例如：
- 發(fā)郵件（用戶不需要等發(fā)完郵件才得到響應）
- 寫日志（用戶不需要日志寫完才看到結果）
- 異步清理操作、緩存刷新、埋點統(tǒng)計等
如何使用 BackgroundTasks？
基本用法：
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def write_log(message: str):
with open("log.txt", "a") as f:
f.write(f"{message}\n")
@app.post("/log")
async def log_operation(background_tasks: BackgroundTasks):
background_tasks.add_task(write_log, "用戶訪問了 /log 接口")
return {"message": "操作已記錄（異步執(zhí)行）"}參數(shù)說明：
參數(shù) | 說明 |
BackgroundTasks | FastAPI 自動注入的后臺任務對象 |
add_task(func, *args, **kwargs) | 注冊后臺執(zhí)行的函數(shù)和參數(shù) |
你的函數(shù)（非 async） | 必須是普通函數(shù)（同步），不能是 async 函數(shù)！ |
為什么函數(shù)不能是 async 的？
因為 FastAPI 的 BackgroundTasks 是基于 Starlette 的同步回調(diào)機制，它在響應后通過事件循環(huán)注冊任務，并同步執(zhí)行你傳入的函數(shù)。
如果你寫成 async def，它不會被 await，也不會進入事件循環(huán)，最終無法運行。
如果我一定要運行 async 函數(shù)怎么辦？
可以使用 Python 的 asyncio.create_task() 方法間接執(zhí)行：
import asyncio
asyncdefasync_job():
print("異步任務開始")
await asyncio.sleep(2)
print("異步任務結束")
defwrapper_async_job():
asyncio.create_task(async_job()) # 間接運行 async 函數(shù)
@app.get("/run")
asyncdefrun(background_tasks: BackgroundTasks):
background_tasks.add_task(wrapper_async_job)
return {"msg": "異步任務已觸發(fā)"}這樣就能兼顧 async def 與 BackgroundTasks！
真實案例：注冊后異步發(fā)送郵件
文件結構如下：
project/
├── main.py
└── tasks/
└── email.pytasks/email.py：
import time
def send_welcome_email(email: str):
time.sleep(3) # 模擬耗時任務
print(f"?? 歡迎郵件已發(fā)送到 {email}")main.py：
from fastapi import FastAPI, BackgroundTasks
from tasks.email import send_welcome_email
app = FastAPI()
@app.post("/register")
async def register(email: str, background_tasks: BackgroundTasks):
# 假設已完成注冊流程
background_tasks.add_task(send_welcome_email, email)
return {"msg": "注冊成功，歡迎郵件將在后臺發(fā)送"}用戶幾乎瞬間得到響應，而郵件將在后臺異步執(zhí)行，提升體驗。
支持多個后臺任務？
當然可以！只需多次調(diào)用 add_task()：
background_tasks.add_task(write_log, "用戶注冊")
background_tasks.add_task(send_welcome_email, email)
background_tasks.add_task(refresh_cache, user_id)它們會按順序在響應后依次執(zhí)行。
與 Depends 依賴注入結合使用
你可以在依賴函數(shù)里注入 BackgroundTasks，實現(xiàn)邏輯更解耦：
def audit(background_tasks: BackgroundTasks):
background_tasks.add_task(write_log, "訪問審計")
return
@app.get("/data")
async def get_data(dep=Depends(audit)):
return {"msg": "數(shù)據(jù)已獲取"}更高級的場景中，你可以把審計、埋點、通知等邏輯都封裝進 Depends，主業(yè)務邏輯保持干凈清晰。
BackgroundTasks 適合什么場景？
適合：
場景 | 示例 |
郵件通知 | 注冊成功、忘記密碼等 |
日志記錄 | 用戶行為埋點、接口訪問日志 |
文件清理 | 上傳后刪除舊文件 |
緩存更新 | 數(shù)據(jù)變動后刷新緩存 |
不適合：
- 需要 重試、隊列控制、任務狀態(tài)追蹤 的任務
- 耗時非常長、可能失敗的操作
對此建議使用 Celery、Dramatiq、APScheduler 等專業(yè)方案。
多任務并發(fā)支持嗎？
支持，但有限制：
- 每個請求的任務是串行執(zhí)行的，不并發(fā)
- 如果任務內(nèi)部使用了 asyncio.create_task()，可以實現(xiàn)并發(fā)邏輯
- 要實現(xiàn)更強大的并發(fā)處理，建議交給專門的任務系統(tǒng)處理（如 Celery）
與 Celery 的對比
特性 | BackgroundTasks | Celery |
部署復雜度 | 簡單，無需額外組件 | 依賴 Redis/RabbitMQ |
是否異步 | ? | ? |
重試機制 | ? 無 | ? 自動重試 |
任務隊列 | ? 無 | ? 內(nèi)置隊列系統(tǒng) |
狀態(tài)追蹤 | ? 無 | ? 支持狀態(tài)與監(jiān)控 |
適用場景 | 簡單后臺邏輯 | 復雜異步任務 |
總結
FastAPI 的 BackgroundTasks 非常適合：
- 小而輕的異步任務（如發(fā)郵件、寫日志）
- 不要求任務可追蹤、重試的場景
- 快速開發(fā)原型、簡潔易用的后端服務
