FastAPI – 非同步程式設計（Async Programming）

async 路由 vs sync 路由差異

---

簡介

在現代 Web 應用程式中，效能與可擴展性往往是決定系統能否成功的關鍵因素。FastAPI 之所以受到廣大開發者喜愛，除了自動產生 OpenAPI 文件、型別檢查外，最重要的特點之一就是對 非同步 (async) 程式設計 的原生支援。

當我們在 FastAPI 中撰寫路由時，可以選擇 同步 (sync) 函式 或 非同步 (async) 函式。兩者在執行流程、資源使用以及錯誤處理上都有顯著差異。了解這些差異不僅能寫出更快的 API，還能避免在高併發環境下出現「阻塞」的致命問題。

本篇文章將從概念、範例、常見陷阱與最佳實踐，逐步說明 async 路由 與 sync 路由 的差別，並提供實務上可直接套用的程式碼範例，協助你在 FastAPI 專案中做出最適合的選擇。

---

核心概念

1. 同步 (sync) 與非同步 (async) 的基本差別

觀點	同步 (sync)	非同步 (async)
執行方式	呼叫者會 阻塞，直到函式執行完成才繼續往下走。	呼叫者會 釋放 CPU，允許其他協程 (coroutine) 同時執行。
底層機制	依賴 OS 執行緒 (Thread) 或進程 (Process)。	依賴 事件迴圈 (event loop) 與 協程 (coroutine)。
適用情境	CPU 密集型、阻塞 I/O 已被封裝為同步函式。	I/O 密集型、需要大量等待（如 DB、HTTP、檔案 I/O）。
資源消耗	每個請求若使用執行緒，會佔用較多記憶體與上下文切換成本。	同一事件迴圈內可同時處理上千個請求，記憶體開銷低。

重點：在 FastAPI 中，若路由函式宣告為 async def，FastAPI 會把它交給 Starlette 的事件迴圈執行；若使用 def，則會在工作執行緒池 (thread pool) 中執行。

2. 為什麼要使用 async 路由？

1. 提升併發量：在高併發環境（如大量同時呼叫外部 API、資料庫查詢）時，async 可以讓單一工作者同時處理多個請求，避免因等待 I/O 而浪費 CPU 時間。
2. 降低資源成本：相較於傳統的多執行緒模型，async 只需要少量的執行緒即可支援大量請求，減少記憶體與上下文切換開銷。
3. 更好的使用者體驗：非同步端點在等待遠端服務回應時不會阻塞其他請求，減少 API 的平均回應時間 (latency)。

3. 何時仍需要 sync 路由？

• CPU 密集型運算（如影像處理、加密解密）不適合放在 async 協程內，應交給背景工作者（Celery、RQ）或使用多執行緒。
• 第三方函式庫僅提供同步介面且無法改寫為 async 時，直接使用 sync 路由較安全。
• 簡單的測試或原型，若不涉及大量 I/O，使用 sync 可以減少開發複雜度。

4. 事件迴圈與執行緒池的互動

FastAPI 內部使用 Starlette 作為底層框架，Starlette 會在接收到請求時檢查路由是 async 還是 def：

• async：直接在事件迴圈中執行 coroutine。
• def：透過 run_in_threadpool 把同步函式委派給 執行緒池，等同於 await run_in_threadpool(sync_func, *args)。

這樣的設計讓開發者可以在同一個應用程式中自由混用 sync 與 async，僅需注意不要在 async 函式裡直接呼叫阻塞的同步程式碼（會「阻塞事件迴圈」）。

---

程式碼範例

以下示範 5 個常見情境，說明 sync 與 async 路由的寫法與差異。所有範例均以 FastAPI 0.110+ 為前提。

範例 1：簡單的同步路由

# file: main_sync.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/sync/hello")
def hello_sync(name: str = "World"):
    """
    同步路由範例：直接回傳字串。
    這類簡單的 CPU 計算不會造成阻塞，可直接使用 sync。
    """
    return {"message": f"Hello, {name}!"}

說明：此路由僅執行基本字串拼接，使用 def 即可。對於 CPU 輕量的工作，sync 與 async 差別不大。

---

範例 2：非同步呼叫外部 API

# file: main_async.py
import httpx
from fastapi import FastAPI

app = FastAPI()

@app.get("/async/external")
async def fetch_weather(city: str = "Taipei"):
    """
    非同步路由範例：使用 httpx.AsyncClient 呼叫第三方天氣 API。
    由於是 I/O 密集型，使用 async 可以在等待回應時處理其他請求。
    """
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"https://api.weather.com/v3/weather/conditions?city={city}")
        data = resp.json()
    return {"city": city, "weather": data.get("weatherDescription")}

重點：httpx.AsyncClient 必須在 await 前使用，否則會阻塞事件迴圈。

---

範例 3：同步資料庫查詢（使用 SQLAlchemy 同步版）

# file: main_db_sync.py
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from .database import SessionLocal, engine, Base
from . import models

Base.metadata.create_all(bind=engine)

app = FastAPI()

def get_db() -> Session:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/sync/users")
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    """
    同步路由：使用傳統的 SQLAlchemy ORM 進行資料庫查詢。
    由於 ORM 為阻塞式，FastAPI 會自動把此函式放入執行緒池。
    """
    users = db.query(models.User).offset(skip).limit(limit).all()
    return {"users": [user.as_dict() for user in users]}

提示：即使路由是 def，FastAPI 仍會在背景執行緒中執行，開發者不必自行管理執行緒。

---

範例 4：非同步資料庫查詢（使用 SQLModel / asyncpg）

# file: main_db_async.py
import asyncio
from fastapi import FastAPI
from sqlmodel import SQLModel, select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/testdb"
engine = create_async_engine(DATABASE_URL, echo=False)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

app = FastAPI()

class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str

@app.on_event("startup")
async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

@app.get("/async/users")
async def read_users(skip: int = 0, limit: int = 10):
    """
    非同步路由：使用 async SQLAlchemy (asyncpg) 直接在協程中執行查詢。
    完全不會阻塞事件迴圈，適合高併發的讀取操作。
    """
    async with async_session() as session:
        stmt = select(User).offset(skip).limit(limit)
        result = await session.exec(stmt)
        users = result.all()
    return {"users": [user.dict() for user in users]}

關鍵：所有與資料庫互動的 I/O 必須使用 await，否則會退回同步行為。

---

範例 5：在 async 路由中誤用阻塞函式（常見陷阱示範）

# file: main_mistake.py
import time
from fastapi import FastAPI

app = FastAPI()

@app.get("/async/bad-sleep")
async def bad_sleep(seconds: int = 5):
    """
    錯誤示範：在 async 路由裡直接使用 time.sleep（阻塞）。
    這會阻塞整個事件迴圈，導致其他請求全部卡住。
    """
    time.sleep(seconds)      # ❌ 阻塞！應改用 asyncio.sleep
    return {"message": f"Slept for {seconds} seconds"}

修正方式：

import asyncio

@app.get("/async/good-sleep")
async def good_sleep(seconds: int = 5):
    await asyncio.sleep(seconds)   # ✅ 非阻塞
    return {"message": f"Slept for {seconds} seconds"}

---

常見陷阱與最佳實踐

陷阱	說明	解決方案
在 async 路由中呼叫阻塞 I/O	如 time.sleep、同步的 requests、阻塞的資料庫驅動。會讓整個事件迴圈卡住。	改用 await asyncio.sleep、httpx.AsyncClient、或將同步呼叫包在 run_in_threadpool 中 (await anyio.to_thread.run_sync(sync_func, ...))。
忘記 await	直接回傳 coroutine 物件，FastAPI 會把它視為回傳值，產生奇怪的 JSON。	確保所有非同步呼叫前都有 await，或使用 asyncio.create_task 若需要背景任務。
混用阻塞與非阻塞資源	同一個路由同時使用 async DB、sync 檔案 I/O，會產生不一致的執行緒行為。	儘量保持同一路由內部的 I/O 風格一致；若必須混用，使用 run_in_threadpool 包裝阻塞部分。
過度使用 async	把所有路由都寫成 async，卻只做 CPU 計算，會增加程式碼複雜度而無效益。	針對 I/O 密集型路由使用 async，CPU 密集型交給背景工作者或保留 sync。
忘記在 startup/shutdown 使用 async	初始化資源時使用同步函式，會阻塞服務啟動。	使用 async def 搭配 await 進行非同步初始化（如資料庫連線池）。

最佳實踐清單

1. 先判斷 I/O 性質：是否涉及網路、磁碟、資料庫等等待操作？若是，優先使用 async。
2. 使用 httpx.AsyncClient、asyncpg、aioredis 等原生 async 客戶端。
3. 避免在 async 函式內直接呼叫阻塞函式，若無法避免，使用 await anyio.to_thread.run_sync(...) 包裝。
4. 把 CPU 密集型任務交給背景工作者（Celery、RQ、Dramatiq），保持 API 回應迅速。
5. 統一錯誤處理：使用 FastAPI 的全域例外處理器 (@app.exception_handler) 來捕捉 asyncio.TimeoutError、httpx.HTTPError 等非同步例外。
6. 測試併發：使用 locust、hey 或 wrk 等工具模擬高併發，觀察同步與非同步路由的延遲與吞吐量差異。

---

實際應用場景

場景	建議使用	為什麼
即時聊天或推播服務	async 路由 + WebSocket	需要同時維持大量長連線，非同步可以有效管理事件迴圈。
第三方支付 API 呼叫	async 路由	多數支付平台回應時間不固定，使用 async 可避免阻塞其他交易。
資料報表產生（大量 DB 聚合）	sync 路由或背景工作者	聚合查詢往往 CPU 密集，建議交給 Celery，API 只回傳任務 ID。
圖像上傳與處理	sync 路由 + 背景工作者	圖像壓縮/辨識屬於 CPU 密集，應該在工作者中處理，API 只負責接收檔案。
IoT 裝置資料收集	async 路由 + async DB (TimescaleDB)	裝置頻繁上報資料，使用 async DB 可以在同一事件迴圈內快速寫入。
簡單的健康檢查 (health check)	sync 路由	執行極簡的 CPU 判斷，無需額外的 async 開銷。

---

總結

• 非同步路由 讓 FastAPI 在 I/O 密集型工作下能夠 高效併發，降低資源使用，同時提升整體吞吐量。
• 同步路由 仍然是處理 CPU 密集 或 只能使用阻塞函式庫 時的安全選擇，FastAPI 會自動把它交給執行緒池，避免阻塞主事件迴圈。
• 關鍵在於判斷：先分析你的端點是屬於 I/O 還是 CPU，根據需求選擇適當的實作方式，並遵守「不要在 async 中阻塞」的原則。
• 透過 正確的 async 客戶端、適當的錯誤處理、以及 背景工作者 的協同運作，你的 FastAPI 服務將能在高流量環境下保持穩定、快速回應。

掌握了 async vs sync 的差異後，你就能在設計 API 時更有底氣，根據實際需求選擇最適合的實作，讓專案在效能與可維護性之間取得最佳平衡。祝你開發順利，打造出高效、彈性的 FastAPI 應用！