非同步
FastAPI 跑在 ASGI 伺服器上,所以路徑函式可以是 def 或 async def。選錯不會立刻壞掉,但在高流量或大量 I/O 時,差別會出現在延遲與吞吐量。
先記一條規則
- 函式裡要
await(例如httpx.AsyncClient、async 資料庫驅動)→ 用async def。 - 函式裡只有同步、會阻塞的工作(一般
time.sleep、同步 SQLAlchemy、CPU 密集計算)→ 用普通def。
FastAPI 會把普通 def 丟到執行緒池,避免卡住事件迴圈。async def 則直接跑在事件迴圈上——這也代表你不能在 async def 裡做長時間的同步阻塞,否則同時間其他請求都會等。
import httpx
from fastapi import FastAPI
app = FastAPI()
@app.get("/price")
async def read_price():
async with httpx.AsyncClient() as client:
response = await client.get("https://httpbin.org/get")
return response.json()
不確定差異時,官方文件的建議是:先寫 def,等你真的需要 await 再改。這比「全部改 async 卻呼叫同步 ORM」安全。
為什麼 I/O 適合 async
API 多數時間在等:等資料庫、等外部 HTTP、等磁碟。await 讓出 CPU,事件迴圈可以去處理別的請求。這就是 FastAPI 能接近 Go / Node 效能數字的原因之一。
不適合硬改 async 的:影像處理、大量加密、純 CPU 迴圈。那些該丟背景工作或獨立 worker,而不是佔著事件迴圈。
BackgroundTasks 與真正的佇列
回應前要寄信、寫稽核紀錄,但不想讓使用者等,可用內建的 BackgroundTasks:
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", "a", encoding="utf-8") as file:
file.write(message + "\n")
@app.post("/notify/")
def notify(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_log, f"notify {email}")
return {"accepted": True}
它適合「短、失敗也不毀滅」的工作。寄給數千人、要重試、要監控的任務,請改用 Celery、ARQ、RQ 這類工作佇列,不要塞進 BackgroundTasks。
進一步學習
- 官方:並行與 async
- 官方:背景工作
- 官方:很趕時間?(async 速讀)
- httpx:同步 / 非同步 HTTP 用戶端
- 下一頁:安全性入門