iT邦幫忙

2025 iThome 鐵人賽

DAY 15
0
Software Development

用 FastAPI 打造你的 AI 服務系列 第 15

[Day 15] 任務管理 (一):Background Task

  • 分享至 

  • xImage
  •  

在 AI 服務中,我們經常遇到一些不需要立即回應給用戶的任務,如推理結果的後處理、日誌記錄、或者資料清理等。FastAPI 的 Background Tasks 功能為這類需求提供了優雅的解決方案,讓我們能夠快速回應用戶,同時在背景處理耗時的輔助工作。

在開發 Web 應用時,我們經常遇到一些不想讓使用者在線上等待的耗時操作。最經典的例子就是使用者註冊成功後,系統需要發送一封歡迎郵件。如果將發信的程式直接寫在 API 路由中,使用者就必須等到信件發送完成才能收到「註冊成功」的回應,這會帶來很差的使用者體驗。

為了解決這個問題,FastAPI 提供了一個極度輕量、無需任何額外設定的內建工具:BackgroundTasks

什麼是 BackgroundTasks?

BackgroundTasks 是一個透過 FastAPI 的 dependency injection (依賴注入) 系統傳入的工具。它允許你在 API 回應 (Response) 已經回傳給使用者之後,才在背景執行一個或多個函式。

它的核心特性是「即發即棄」(fire-and-forget)。也就是說,你的主程式在提交任務後,就不會再關心它是否成功、何時完成,而是立刻繼續向下執行,最終回傳 HTTP 回應。

基礎程式碼範例

要使用它非常簡單,只需要在你的路由函式參數中,宣告一個 BackgroundTasks 型別的變數即可。

from fastapi import FastAPI, BackgroundTasks
import time

app = FastAPI()

def write_log(message: str):
    """一個模擬寫入日誌的慢速函式"""
    time.sleep(5)  # 模擬 I/O 延遲
    with open("log.txt", mode="a") as log_file:
        log_file.write(message)
        print(f"任務完成: {message.strip()}")

@app.post("/send-notification/{email}")
def send_notification(email: str, background_tasks: BackgroundTasks):
    message = f"通知信已發送給 {email}\n"
    # 使用 .add_task() 將函式與其參數加入背景任務佇列
    background_tasks.add_task(write_log, message)
    # 這行會立刻回傳,使用者無需等待 5 秒
    return {"message": "正在背景發送通知,請稍候..."}

當你呼叫這個 API 時,會立刻收到 JSON 回應。與此同時,伺服器端的終端機會在 5 秒後印出 "任務完成" 的訊息,並且 log.txt 檔案會被建立與寫入。

此外,在執行 BackgroundTasks 的這 5 秒期間,FastAPI 是可以處理其他 Request 的。

讓我們稍微調整一下程式碼,方便進行展示:

+ import logging
import time
from datetime import datetime

from fastapi import BackgroundTasks, FastAPI

+ # 設定 logger
+ logger = logging.getLogger("uvicorn")

app = FastAPI()


@app.get("/")
def read_root():
+   current_time = datetime.now().strftime("%H:%M:%S")
+   logger.info(f"[{current_time}] 讀取根目錄")
    return {"message": "Hello World"}


def write_log(message: str):
    """一個模擬寫入日誌的慢速函式"""
    time.sleep(5)  # 模擬 I/O 延遲
    with open("log.txt", mode="a", encoding="utf-8") as log_file:
        log_file.write(message)
+       current_time = datetime.now().strftime("%H:%M:%S")
+       logger.info(f"[{current_time}] 任務完成: {message.strip()}")

@app.post("/send-notification/{email}")
def send_notification(email: str, background_tasks: BackgroundTasks):
    message = f"通知信已發送給 {email}\n"
+   current_time = datetime.now().strftime("%H:%M:%S")
+   logger.info(f"[{current_time}] {message.strip()}")
    background_tasks.add_task(write_log, message)
    return {"message": "正在背景發送通知,請稍候..."}

在發出 /send-notification/{email} API 請求後的 5 秒內,又打了 3 次 / API,都可以正常取得回應~

優點與侷限

優點:

  • 簡單快速:無需安裝或設定任何額外套件。
  • 無縫整合:與 FastAPI 的相依性注入完美結合。

侷限:

  • 不可靠:如果執行任務的過程中,伺服器重啟或崩潰,這個任務就會永遠遺失。
  • 無法追蹤:你無法得知任務的執行狀態、成功與否或回傳值。
  • 資源共享:背景任務與你的 Web 應用在同一個程序 (Process) 中執行,如果任務是 CPU 密集型的,可能會影響 API 的回應效能。

小結

BackgroundTasks 是處理輕量、非關鍵性任務的絕佳選擇,例如記錄日誌、發送不重要的通知等。但如果你的任務至關重要,不容許失敗,或是需要更複雜的管理,那麼我們就需要更專業的工具了。下一篇文章,我們將探討如何處理另一種常見的挑戰:同步阻塞任務。


上一篇
[Day 14] 資料庫
系列文
用 FastAPI 打造你的 AI 服務15
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言