Web Day 19 背景任務:FastAPI BackgroundTasks 與 Celery 入門
執行需求:CPU 可跑;Celery 部分需 Redis(如無 Redis 可改用 BackgroundTasks 或 APScheduler)。今天是「Web 系統實戰:用 FastAPI 打造能上線的後端」系列的第十九篇。我們已經能在端點裡做認證、查資料庫、回應 JSON;但真實世界的 API 經常需要「使用者按了按鈕、伺服器立刻回 200、實際工作慢慢做」這種模式。今天的主題就是把這個模式拆開:什麼時候用 FastAPI 內建的 BackgroundTasks 就好、什麼時候必須上 Celery 5.5、Windows 與本機開發的替代方案是什麼。
引言
「背景任務」是一個很常被誤會的詞。直覺上你可能會想:「不就丟到 thread pool 嗎?」沒錯,BackgroundTasks 的確是在 thread pool 裡跑;但真正關鍵的問題是「任務能不能失敗重試、能不能追蹤狀態、能不能被外部系統看到」。當你只需要「回應後順手做一件事」,BackgroundTasks 就夠了;當你需要「任務在另一台機器上執行、要能重試三次失敗、要能從管理介面看到任務狀態」,那就該上 Celery 或 RQ 之類的工作佇列。
今天的範例同樣接續 Day 16-18 的專案。我們會做四件事:第一,理解 BackgroundTasks 的工作模型與限制;第二,寫一個完整可執行的 BackgroundTasks 範例;第三,介紹 Celery 5.5 的最小設定並說明 Windows 與小型服務的替代方案;第四,整理常見踩雷,包括「任務跑很久 client 卻 timeout」、「任務丟出去就再也找不回來」。
BackgroundTasks:FastAPI 內建的輕量方案
BackgroundTasks 的 API 很簡潔:把函式與參數丟進 background.add_task,FastAPI 會在送出回應之後再呼叫那個函式。函式可以在 async 端點裡跑、也可以是同步函式;同步函式會被丟到 anyio 的 thread pool。
# app/main.py
# BackgroundTasks 第一個範例:使用者建立預約後寄通知(模擬)
import time
from datetime import datetime
from fastapi import BackgroundTasks, FastAPI
from pydantic import BaseModel
app = FastAPI(title="Web Day 19 範例", version="0.19.0")
class Booking(BaseModel):
user_id: int
service: str
scheduled_at: datetime
def send_notification(user_id: int, message: str) -> None:
# 模擬寄信:真實情境會呼叫 SMTP、SES、Twilio 等等
time.sleep(0.2)
print(f"[通知] user={user_id} {message} at {datetime.now().isoformat()}")
@app.post("/bookings", status_code=201)
def create_booking(booking: Booking, background: BackgroundTasks):
# 真正的「建立預約」邏輯通常很短
booking_id = booking.user_id * 1000 + int(booking.scheduled_at.timestamp()) % 1000
# 把「寄通知」丟到背景,端點本身立刻回應
background.add_task(
send_notification,
booking.user_id,
f"您的 {booking.service} 預約已成立,編號 {booking_id}",
)
return {"booking_id": booking_id, "status": "created"}
這個端點做了兩件事:「建立預約」是同步邏輯,回傳 booking_id;「寄通知」丟到背景,使用者不會等到通知寄出才收到回應。background.add_task(func, *args, **kwargs) 會把 func(*args, **kwargs) 排進背景佇列;FastAPI 在送出 HTTP 回應後才執行這個任務。
BackgroundTasks 有幾個重要的限制你要記得:
- 任務與 request 共用同一個行程:FastAPI 行程重啟、當機、scale down 任務就會消失。
- 任務沒有重試機制:寫檔失敗、第三方 API timeout,這次就沒了。
- 沒有管理介面:想知道「昨天寄了幾封信」只能去翻 log。
- 沒有優先級:所有任務都跑在同一個 thread pool。
如果你的任務能容忍「偶爾漏一兩件」、能在「行程死掉時整批重來」、不需要從後台查狀態,BackgroundTasks 就是最快、最簡單的選擇。今天範例的「寄通知」剛好符合這個特性,所以我們用它就好。
完整實作:可測試的 BackgroundTasks
要測 BackgroundTasks 的端點,比測同步端點多了一件事:要驗證「背景函式真的被呼叫」。FastAPI 的 TestClient 提供一個 task 屬性,能讓你在測試裡看到排進去的任務:
# tests/test_background.py
# 用 TestClient 驗證 BackgroundTasks
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_create_booking_runs_background_task(monkeypatch):
calls = []
def fake_send_notification(user_id: int, message: str):
calls.append((user_id, message))
# 把背景函式換成假實作,避免真的 print / sleep
monkeypatch.setattr("app.main.send_notification", fake_send_notification)
response = client.post(
"/bookings",
json={"user_id": 42, "service": "攝影", "scheduled_at": "2025-08-15T10:00:00"},
)
assert response.status_code == 201
body = response.json()
assert body["status"] == "created"
# 驗證背景任務真的被排入
assert len(client.task) == 1
# 把任務真的執行一次(測試期間可以這樣做)
client.task[0].run()
assert calls == [(42, f"您的 攝影 預約已成立,編號 {body['booking_id']}")]
def test_create_booking_does_not_wait_for_slow_notification(monkeypatch):
def slow_send_notification(user_id: int, message: str):
import time
time.sleep(2.0) # 故意做慢一點
monkeypatch.setattr("app.main.send_notification", slow_send_notification)
# 用時間戳確認端點本身沒有等背景任務
import time
start = time.perf_counter()
response = client.post(
"/bookings",
json={"user_id": 7, "service": "諮詢", "scheduled_at": "2025-08-15T11:00:00"},
)
elapsed = time.perf_counter() - start
assert response.status_code == 201
# 端點本身應該遠小於 2 秒
assert elapsed < 1.0
第一個測試透過 client.task 取得 FastAPI 排入的背景任務,用 task.run() 在測試裡真的執行它,再用 spy 變數驗證呼叫了什麼。第二個測試則是反向驗證:故意把背景函式做慢,確認端點本身不會被卡住。這兩個測試合在一起,就驗證了「BackgroundTasks 的兩個關鍵特性」:任務確實會跑、任務不會擋住主回應。
另外一個寫法是把背景任務做成「可注入的依賴」,這樣測試可以完全換掉它:
# app/notify.py
# 把通知邏輯抽成一個介面,方便測試與日後換實作
from typing import Protocol
class Notifier(Protocol):
def send(self, user_id: int, message: str) -> None: ...
class PrintNotifier:
def send(self, user_id: int, message: str) -> None:
print(f"[通知] user={user_id} {message}")
def get_notifier() -> Notifier:
# 真正的實作應該是 SMTP/SES/Twilio,這裡先用 print
return PrintNotifier()
# app/main.py(用依賴注入版本)
from fastapi import BackgroundTasks, Depends, FastAPI
from pydantic import BaseModel
from app.notify import Notifier, get_notifier
app = FastAPI(title="Web Day 19 範例", version="0.19.0")
class Booking(BaseModel):
user_id: int
service: str
@app.post("/bookings", status_code=201)
def create_booking(
booking: Booking,
background: BackgroundTasks,
notifier: Notifier = Depends(get_notifier),
):
booking_id = booking.user_id * 1000
background.add_task(notifier.send, booking.user_id, f"預約 {booking.service} 成立")
return {"booking_id": booking_id}
這樣做的好處是「測試可以換掉 get_notifier 來換掉 notifier」,背景任務的入口仍然維持不變,程式碼的可演進性比較好。Day 17 的 app.dependency_overrides 正是這個寫法的最佳搭配。
Celery 5.5:真的工作佇列
當任務需要「可靠的傳遞」、「失敗重試」、「狀態追蹤」、「跨行程執行」時,BackgroundTasks 就不夠了。Celery 是 Python 生態最成熟的工作佇列框架,2025 年的主流版本是 5.5 世代。它的基本元件有三個:
- Task:定義一個要被背景執行的函式。
- Broker:存放任務的訊息中介,Redis 或 RabbitMQ 是常見選擇。
- Worker:跑任務的行程,可以有多個、可以分散在不同機器。
我們先安裝:
# 命令列:安裝 Celery 與 Redis client
uv add "celery==5.5"
uv add "redis==5.2"
# 啟動 Redis 容器(Docker)或本機 Redis
docker run -d -p 6379:6379 --name web-redis redis:8-alpine
# 或在 Windows 本機裝 Redis 8 之後直接執行 redis-server
接著定義 Celery 應用:
# app/celery_app.py
# Celery 應用:設定 broker 與結果後端
from celery import Celery
celery_app = Celery(
"web_day_19",
broker="redis://localhost:6379/0",
backend="redis://localhost:6379/1",
include=["app.tasks"],
)
celery_app.conf.update(
task_serializer="json",
accept_content=["json"],
result_serializer="json",
timezone="Asia/Taipei",
enable_utc=True,
task_track_started=True,
task_acks_late=True,
)
這個檔案建立 Celery 應用物件,指定 Redis 8 為 broker 與 result backend。task_acks_late=True 讓任務執行成功後才確認 (ack),避免 worker 死掉時任務遺失;task_track_started=True 讓我們能看到「任務已開始執行」的狀態。
接著定義任務:
# app/tasks.py
# 背景任務的定義檔
import time
from celery import shared_task
@shared_task(bind=True, max_retries=3, default_retry_delay=10)
def send_notification(self, user_id: int, message: str) -> dict:
try:
# 模擬寄信:真實情境會呼叫 SMTP/SES/Twilio
time.sleep(0.5)
return {"user_id": user_id, "message": message, "status": "sent"}
except Exception as exc:
raise self.retry(exc=exc)
@shared_task 讓這個任務可以被 Celery 自動發現,bind=True 讓函式拿到 self(用於 retry),max_retries=3 限制最多重試三次,default_retry_delay=10 設定重試間隔。當 send_notification 拋例外時,Celery 會自動排程重試,這是 BackgroundTasks 完全做不到的事。
最後在端點裡呼叫它:
# app/main.py(Celery 版本)
from fastapi import FastAPI
from pydantic import BaseModel
from app.tasks import send_notification
app = FastAPI(title="Web Day 19 Celery 範例", version="0.19.0")
class Booking(BaseModel):
user_id: int
service: str
@app.post("/bookings", status_code=202)
def create_booking(booking: Booking):
# 把任務丟到 broker;端點本身不等待
async_result = send_notification.delay(booking.user_id, f"預約 {booking.service} 成立")
return {
"task_id": async_result.id,
"status": "queued",
}
@app.get("/tasks/{task_id}")
def get_task_status(task_id: str):
# 查詢任務狀態(這就是 BackgroundTasks 做不到的事)
result = send_notification.AsyncResult(task_id)
return {
"task_id": task_id,
"state": result.state,
"ready": result.ready(),
"result": result.result if result.ready() else None,
}
這個版本用 send_notification.delay(...) 把訊息推進 Redis;端點立刻回 202 Accepted。AsyncResult(task_id) 讓我們查詢任務狀態(queued、started、success、failure),這對「任務跑了沒」的可觀察性是巨大的提升。啟動 Celery worker 的命令列指令是:
# 命令列:啟動 Celery worker
uv run celery -A app.celery_app worker --loglevel=info --concurrency=2
# 預期輸出(節錄):
# [tasks]
# . app.tasks.send_notification
# [2025-08-04 10:00:00,000: INFO/MainProcess] celery@host ready.
有了 worker 之後,端點丟出去的任務會被 worker 接走、執行、寫回結果;查詢端點 /tasks/{task_id} 能即時看到任務進度。
Windows 與小型服務的替代方案
在 Windows 與本機開發環境,Celery + Redis 不見得是首選。原因是 Celery 在 Windows 上的 fork 行為有限制(prefork pool 不可用,需用 solo 或 threads pool);本機小服務可能也不需要 broker。本節列出三個替代方案,按規模遞增:
方案一:BackgroundTasks。前面示範過,零基礎設施、最簡單。適合「任務能容忍偶爾漏」、規模在數十 req/s 以內的場景。整個 FastAPI 行程死掉、任務就會一起消失,這是它的限制也是它的簡單。
方案二:APScheduler(Web Day 21 詳細介紹)。當你需要「每五分鐘跑一次」或「延遲 30 秒後跑」,APScheduler 3.11.x 可以把任務排進記憶體排程。沒有 broker、不需要額外行程,純粹是 FastAPI 行程裡的 scheduler。適合「週期性任務 + 偶爾丟一個延遲任務」。
方案三:Celery + Redis(推薦)。當任務需要「可靠的傳遞、失敗重試、狀態追蹤、跨行程、跨機器」,Celery 是最成熟的選擇。Windows 上把 worker 換成 solo 或 threads pool 也行;正式上線再切 Linux + prefork。
實務上的判斷流程是這樣的:
- 任務是「使用者動作觸發的小事」(寄通知、寫審計 log)?用 BackgroundTasks。
- 任務是「延遲或週期性的事」(30 秒後寄提醒、每小時清暫存)?用 APScheduler。
- 任務是「耗時、可能失敗、需要狀態查詢、需要跨行程」?用 Celery。
很多小型服務根本用不到 Celery。Web Day 35-44 的預約管理系統專案裡,「寄提醒信」用 BackgroundTasks、「每小時清空過期預約」用 APScheduler;只有「批次處理大量資料」才會升級到 Celery。
三種方案的橫向比較:選哪一個
為了讓你在不同階段都能選對工具,我把三種方案放進同一張表:
| 特性 | BackgroundTasks | APScheduler | Celery 5.5 + Redis |
|---|---|---|---|
| 額外行程 | 不需要 | 不需要 | 需要 worker |
| 外部 broker | 不需要 | 不需要 | 需要 Redis 或 RabbitMQ |
| 失敗重試 | 無 | 有限(用 misfire 設定) | 完整支援 |
| 狀態查詢 | 無 | 有限(job store) | AsyncResult 完整查詢 |
| 跨機器 | 不行 | 不行 | 原生支援 |
| 適合規模 | 小型、低 QPS | 週期任務、延遲任務 | 中大型、需可靠傳遞 |
實務上最常見的錯誤是「為了將來可能的規模,把小服務直接架上 Celery」。Celery 的 broker 設定、worker 管理、任務監控,每一項都是額外的營運成本。如果你的服務目前每日只有數千次背景任務呼叫,BackgroundTasks 加上 APScheduler 就能撐很久;等流量真的上來、任務真的開始掉,再升級到 Celery 也來得及。
常見錯誤與踩雷
第一個雷是「BackgroundTasks 用了 async 函式卻沒注意」。BackgroundTasks 的設計是「背景函式跑在 thread pool」,所以 async def 函式丟進去會被警告或直接忽略。FastAPI 0.116 對這個情況會印出 RuntimeWarning: coroutine was never awaited,但不會中斷程式。解法是「BackgroundTasks 裡只放 sync 函式;要 await async 函式就改用 lifespan 或任務佇列」。
第二個是「Celery 任務序列化失敗」。預設 Celery 用 pickle 序列化任務參數,當你傳 Pydantic 模型、datetime、Decimal 這類物件時,多機環境下的序列化會丟 PicklingError。解法是在 celery_app.conf 裡把 task_serializer 改成 json,並把參數縮減成基本型別(int、str、float、bool、list、dict)。我們今天的範例已經預先設成 json,這是 2025 年的主流選擇。
第三個是「Celery worker 沒啟動、任務卡在 queued」。這是新人最常見的「為什麼任務都沒跑」原因。記得在背景另外開一個終端機跑 celery -A app.celery_app worker --loglevel=info。CI 環境通常會用 supervisor 或 docker-compose 拉起 worker;單元測試可以用 task_always_eager=True 讓任務同步執行,今天的範例保留 async 模式,正式測試再切。
第四個是「忘記處理任務失敗」。Celery 任務失敗時,預設會把錯誤記到 log,但不會通知應用程式。實務上你會想接 task_failure signal 或在任務內部用 try/except 包起來,把錯誤寫進應用程式的日誌(Web Day 22 會深入)。今天的範例故意展示 retry 機制,正是為了提醒「失敗是常態」。
任務參數的設計原則:傳 ID 而不是傳物件
這條原則不只適用 Celery,所有背景任務都應該這樣設計:把參數設計成「任務自己可以重新從資料庫撈到最新狀態」。舉例來說,「寄通知」任務應該接收 user_id 與 booking_id,而不是整個 User 物件與 Booking 物件。這樣做的好處是:worker 拿到的永遠是最新的資料,避免「使用者改了預約但 worker 用的是舊參數」這種競態。
另外一個常被忽略的細節是「任務參數的大小」。把整個 1 MB 的 JSON 物件塞進 Redis 雖然可行,但會拖慢序列化與網路傳輸。實務上「任務參數 ≤ 1 KB」是個好用的經驗法則;如果真的需要傳大量資料,先把資料寫到物件儲存(S3、MinIO)再傳 URL。今天的範例刻意只傳 user_id 與 message,後續專案篇也會維持這個習慣。
效能與實務提醒
第一個提醒是「BackgroundTasks 適合的延遲範圍」。當任務能在幾百毫秒內完成、又不會被常常呼叫,BackgroundTasks 跟 thread pool 就夠。當任務常常花到 1 秒以上、或 QPS 高到數百,建議直接上 Celery,避免 thread pool 被佔滿。
第二個是「broker 選擇」。Redis 8 是最方便的 broker,但它有一個重要特性:list 結構的任務被消費後就消失,無法做「保留七天稽核」。如果你的任務需要被永久保留(例如審計軌跡),RabbitMQ 的持久化更合適。對「通知、報表、清理」這種場景,Redis 就夠了。
第三個是「任務參數設計」。把參數設計成「任務自己可以重新從資料庫撈到最新狀態」是最安全的做法:傳遞 booking_id 而不是整個 booking 物件,這樣 worker 拿到的永遠是最新的資料,避免「使用者改了預約但 worker 用的是舊參數」這種競態。今天的範例刻意只傳 user_id 與 message 就是這個原因。
第四個是「測試策略」。BackgroundTasks 的測試用 TestClient + monkeypatch 就夠;Celery 的測試則建議用 task_always_eager=True 或 app.conf.task_always_eager = True 在 pytest 啟動時設好,讓任務在測試裡同步執行。Web Day 40「專案:測試與驗收」會在真實專案裡套這個寫法。
小結
今天把「背景任務」這個主題拆成三層:BackgroundTasks(零基礎設施、輕量)、APScheduler(純記憶體排程,Day 21 詳細介紹)、Celery 5.5 + Redis 8(可靠的任務佇列)。FastAPI 端點可以用 background.add_task 處理短任務、用 .delay() 把任務丟給 Celery、用 AsyncResult 查狀態。Windows 與本機開發可以用 BackgroundTasks 或 APScheduler 替代 Celery,避免額外的 broker 設定。我們也示範了「用 TestClient 與 client.task 驗證背景任務真的執行」、「用 monkeypatch 換掉背景函式」這兩個常用測試手法。
結語
今天學會了「把工作丟到背景」,但所有背景任務目前都是「做一次就結束」。明天,Web Day 20「快取:Redis 與快取策略」會把 Redis 從 broker 換成 cache:把同一筆資料暫存起來、讓第二次以後的存取直接命中。今天的最後一段已經把 Redis 8 跑起來了,明天就直接用同樣的環境接快取。
本週回顧:品質區塊前四天的整合
Web Day 16 到 19 是品質區塊的第一段:測試、測試資料、非同步、背景任務。這四篇串起來其實是同一個故事——「怎麼讓 FastAPI 服務在生產環境正確、可靠、快速」。Day 16-17 的 pytest 與 fixture 解決「我怎麼知道改動沒壞」,Day 18 的 async 與 ASGITransport 解決「端點怎麼在事件迴圈裡測」,今天的 BackgroundTasks 與 Celery 解決「重的工作怎麼不擋住使用者」。
這四天的工具有一個共通點:都能在 CPU 環境完整跑起來,不需要昂貴的基礎設施。BackgroundTasks 與 APScheduler 沒有 broker,Celery 可以用同一個 Redis 8 容器,pytest 的 SQLite 記憶體模式完全在本機。這也呼應了 SPEC-web-series 開頭的承諾:整個系列的範例都是「CPU 可跑」,只有少數篇章需要 Docker 跑 Redis 或 PostgreSQL。今天的工作到此告一段落,明天讓我們把 Redis 8 從 broker 換成 cache,看看「第二次以後的存取」能快多少。
如果你這四天都有跟上,現在你的專案結構應該長這樣:app/ 裡有 main、db、celery_app、tasks、notify;tests/ 裡有 conftest、test_main、test_items、test_async_users、test_background。每加一個新端點,就在 tests/ 加一個 test_xxx.py,從 conftest.py 拿 client 與 db_session,5 分鐘就能把測試骨架搭起來。這個節奏會一路延續到 Web Day 40 的專案驗收篇。
延伸資源
- FastAPI BackgroundTasks 官方教學(0.116,2025-07):
https://fastapi.tiangolo.com/tutorial/background-tasks/ - Celery 官方入門(5.5,2025):
https://docs.celeryq.dev/en/stable/getting-started/introduction.html - Redis 官方文件(8,2025):
https://redis.io/docs/latest/ - redis-py 官方文件(5.2,2025):
https://redis.readthedocs.io/en/stable/ - Celery 在 Windows 上的注意事項(2025):
https://docs.celeryq.dev/en/stable/userguide/windows.html
留言
張貼留言