Skip to content

建立 2026-09-14 更新 2026-09-14

非同步

FastAPI 跑在 ASGI 伺服器上,所以路徑函式可以是 defasync 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

進一步學習