FastAPI異步任務:用BackgroundTasks處理耗時操作

在Web開發中,我們經常會遇到需要執行耗時操作的場景,比如發送郵件、生成報表或者處理大量數據。如果直接在API接口中處理這些操作,用戶會被迫等待操作完成才能收到響應,這嚴重影響了用戶體驗。FastAPI爲了解決這個問題,提供了BackgroundTasks工具,讓我們可以在後臺異步處理這些耗時任務,用戶無需等待。

一、爲什麼需要後臺任務?

想象一下,用戶提交一個表單後,你需要給用戶發送一封確認郵件。如果直接在接口中發送郵件(假設發送郵件需要3-5秒),用戶必須等待這幾秒才能看到“提交成功”的提示,這顯然很不友好。而使用BackgroundTasks,接口可以立即返回“任務已開始處理”,郵件會在後臺慢慢發送,用戶體驗會好很多。

二、FastAPI的BackgroundTasks是什麼?

BackgroundTasks是FastAPI提供的一個類,用於管理請求處理完成後纔會執行的後臺任務。它的核心作用是:
- 在API接口返回響應後,自動執行預先註冊的耗時操作。
- 不阻塞用戶等待,提升接口響應速度。

三、如何使用BackgroundTasks?

使用步驟非常簡單,只需3步:

1. 導入BackgroundTasks

fastapi中導入BackgroundTasks類:

from fastapi import FastAPI, BackgroundTasks

2. 在路由函數中聲明BackgroundTasks參數

在需要處理後臺任務的路由函數中,添加一個bg: BackgroundTasks參數(FastAPI會自動注入該參數):

@app.post("/submit")
async def submit_task(bg: BackgroundTasks):
    # 後續操作...

3. 通過add_task添加後臺任務

使用bg.add_task()方法,將需要後臺執行的函數和參數註冊進去。add_task()的第一個參數是函數名,後續參數是函數的參數:

bg.add_task(耗時函數, 參數1, 參數2, ...)

四、代碼示例:模擬後臺任務

我們用一個簡單的例子演示如何用BackgroundTasks處理耗時操作。假設我們需要生成一個大文本文件(模擬耗時操作),用戶提交請求後,接口立即返回成功,文件生成在後臺執行。

完整代碼:

from fastapi import FastAPI, BackgroundTasks
import time

app = FastAPI()

# 模擬耗時操作:生成文本文件(耗時5秒)
def generate_large_file(file_name: str, content: str):
    # 模擬寫入文件的耗時操作
    time.sleep(5)  # 假設寫入文件需要5秒
    with open(file_name, "w", encoding="utf-8") as f:
        f.write(content)
    print(f"文件 {file_name} 已生成成功!")

@app.post("/create-file")
async def create_file_task(
    file_name: str = "test.txt", 
    content: str = "這是一個大文件內容", 
    bg: BackgroundTasks
):
    # 將生成文件的任務添加到後臺
    bg.add_task(generate_large_file, file_name, content)

    # 立即返回響應,告知用戶任務已在後臺處理
    return {
        "message": "文件生成任務已啓動",
        "file_name": file_name,
        "status": "後臺處理中"
    }

五、運行與測試

  1. 運行FastAPI應用(使用Uvicorn):
   uvicorn main:app --reload
  1. 測試接口:
    - 使用Postman或curl發送POST請求到http://localhost:8000/create-file
    - 接口會立即返回響應:{"message": "文件生成任務已啓動", "file_name": "test.txt", "status": "後臺處理中"}
    - 此時可以看到控制檯打印:文件 test.txt 已生成成功!(耗時5秒,與接口響應無關)。

六、關鍵點說明

  1. 任務執行時機BackgroundTasks的任務在接口返回響應後纔會執行,不會阻塞用戶等待。

  2. 參數傳遞add_task()支持位置參數和關鍵字參數,例如:

   # 位置參數
   bg.add_task(generate_large_file, "a.txt", "內容A")

   # 關鍵字參數
   bg.add_task(generate_large_file, file_name="b.txt", content="內容B")
  1. 任務順序:多個任務會按註冊順序執行,例如:
   bg.add_task(task1)
   bg.add_task(task2, param1)

會先執行task1,再執行task2

  1. 適用場景:適合I/O密集型任務(如網絡請求、文件讀寫),不適合CPU密集型任務(如複雜計算),否則可能阻塞其他請求。

七、注意事項

  • 任務失敗處理BackgroundTasks不會捕獲任務執行中的異常,若任務報錯,FastAPI會忽略(需自行處理異常)。
  • 生產環境限制:若FastAPI應用重啓或崩潰,未執行的後臺任務會丟失,因此不適合需要持久化的任務(如訂單處理)。
  • 與異步函數兼容:若任務本身是異步函數(async def),可直接通過add_task()註冊(FastAPI會自動處理異步執行)。

總結

BackgroundTasks是FastAPI中處理後臺耗時任務的輕量工具,通過它可以快速提升接口響應速度,優化用戶體驗。核心用法是:導入類→聲明參數→註冊任務→返回響應。適用於發送郵件、生成文件、日誌記錄等非關鍵路徑的耗時操作,讓接口既能快速響應用戶,又能保證任務完成。

小夜