FastAPI 基礎概念：同步 vs 非同步特性

簡介

在現代 Web 開發中，效能 與 可擴展性 常常是選擇框架的關鍵因素。FastAPI 以 高效能、自動產生 API 文件 為賣點，背後則是基於 Starlette（ASGI）與 Pydantic 的設計。
其中最能體現 FastAPI 優勢的，就是它同時支援 同步（sync） 與 非同步（async） 的路由函式。了解這兩種模式的差異與適用情境，能讓你寫出既易讀又高效的 API。

本篇文章將從概念說明、實作範例、常見陷阱與最佳實踐，乃至於真實的應用場景，完整介紹 FastAPI 的同步與非同步特性，幫助初學者快速上手，同時提供中級開發者進一步優化的方向。

核心概念

1. 為什麼同時支援 sync / async？

• 同步函式（普通的 def）在執行期間會 阻塞 執行緒，適合 CPU‑bound 或不需要等待 I/O 的簡單任務。
• 非同步函式（async def）配合 await 可以在等待 I/O（如資料庫、外部 API、檔案）時釋放執行緒，讓同一個工作者（worker）同時處理多個請求，提高併發度。

FastAPI 讓開發者可以根據每個端點的需求自由選擇，避免「一刀切」的設計，減少不必要的資源浪費。

2. 基本語法差異

同步函式	非同步函式
def read_item(item_id: int):	async def read_item(item_id: int):
直接呼叫阻塞式程式碼	必須使用 await 呼叫支援 await 的 I/O 函式
併發受限於工作者數量	同一工作者可同時處理多個請求（取決於事件迴圈）

注意：即使使用 async def，如果裡面只執行同步阻塞程式碼，仍會造成效能瓶頸。必須搭配支援 await 的庫（如 httpx, databases, aioredis）才能真正發揮非同步優勢。

3. 何時選擇 sync，何時選擇 async？

場景	建議使用	理由
讀寫本地檔案（小檔案）	sync	檔案 I/O 時間短，額外的 await 開銷不划算
呼叫外部 REST API、資料庫查詢	async	這類 I/O 延遲較高，使用 async 可釋放工作者
CPU 密集計算（圖像處理、機器學習）	sync + background task	CPU 任務不適合放在 async 中，建議交給背景工作者或 Celery
快速回傳靜態資料	sync	無 I/O，直接回傳即可

4. 實作範例

4.1 同步端點：回傳簡易訊息

from fastapi import FastAPI

app = FastAPI()

@app.get("/hello")
def hello():
    """
    同步端點，直接回傳字串。
    適合不需要 I/O 的簡單操作。
    """
    return {"message": "Hello, FastAPI!"}

4.2 非同步端點：呼叫外部 API

import httpx
from fastapi import FastAPI

app = FastAPI()

@app.get("/weather/{city}")
async def get_weather(city: str):
    """
    非同步端點，使用 httpx 的 async client 向第三方天氣 API 發送請求。
    在等待回應的同時，事件迴圈會切換去處理其他請求。
    """
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"https://api.example.com/weather/{city}")
        data = resp.json()
    return {"city": city, "temperature": data["temp"], "status": data["status"]}

4.3 非同步資料庫查詢（使用 databases 套件）

import databases
import sqlalchemy
from fastapi import FastAPI

DATABASE_URL = "sqlite+aiosqlite:///./test.db"
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()

users = sqlalchemy.Table(
    "users",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("name", sqlalchemy.String),
)

engine = sqlalchemy.create_engine(
    DATABASE_URL.replace("aiosqlite", "sqlite"), connect_args={"check_same_thread": False}
)
metadata.create_all(engine)

app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

@app.get("/users/{user_id}")
async def read_user(user_id: int):
    """
    非同步查詢單筆使用者資料。
    使用 databases 提供的 awaitable 方法，避免阻塞。
    """
    query = users.select().where(users.c.id == user_id)
    row = await database.fetch_one(query)
    if row is None:
        return {"error": "User not found"}
    return {"id": row["id"], "name": row["name"]}

4.4 同步端點包裝非同步工作（使用 BackgroundTasks）

from fastapi import FastAPI, BackgroundTasks
import time

app = FastAPI()

def write_log(message: str):
    """同步函式，寫入檔案（模擬較慢的 I/O）"""
    with open("log.txt", "a") as f:
        f.write(message + "\n")
    time.sleep(2)  # 模擬阻塞

@app.post("/submit")
async def submit(data: dict, background_tasks: BackgroundTasks):
    """
    主要邏輯使用 async，將耗時的同步寫檔工作交給 BackgroundTasks。
    這樣即使是同步程式，也不會阻塞主請求。
    """
    # 處理主要業務...
    result = {"status": "accepted", "data": data}
    # 加入背景任務
    background_tasks.add_task(write_log, f"Received: {data}")
    return result

4.5 同步端點呼叫非同步函式（不建議）

import asyncio
from fastapi import FastAPI

app = FastAPI()

def sync_wrapper():
    """
    同步函式中直接呼叫 async 函式會產生 RuntimeError，
    必須使用 asyncio.run() 或建立事件迴圈。
    但這樣會阻塞，失去 async 的好處，故不建議這樣寫。
    """
    return asyncio.run(async_task())

async def async_task():
    await asyncio.sleep(1)
    return "Done"

@app.get("/bad-sync")
def bad_sync():
    # 這裡會同步等待 async_task 完成，效能下降
    return {"result": sync_wrapper()}

常見陷阱與最佳實踐

1. 混用 sync/async 造成阻塞

   • 在 async def 中 直接呼叫阻塞函式（例如 requests.get、time.sleep）會讓事件迴圈卡住。
   • 解法：改用支援 await 的庫（httpx.AsyncClient、asyncio.sleep），或將阻塞工作交給 BackgroundTasks / ThreadPoolExecutor。

2. 忘記在啟動/關閉事件中連接/斷開非同步資源

   • 如資料庫、Redis 客戶端等，需要在 @app.on_event("startup") / shutdown 中使用 await 連接。
   • 若忘記，會在第一個請求時才建立連線，造成突發的延遲。

3. 過度使用 async

   • 若端點僅執行 CPU 密集運算，使用 async def 並不會提升效能，反而會增加程式複雜度。
   • 建議：將 CPU 任務搬到 背景工作者（Celery、RQ）或使用 ProcessPoolExecutor。

4. 資料庫連線池配置不當

   • 非同步資料庫套件（如 databases、SQLModel）需要適當的 max_connections，否則高併發時會出現連線耗盡。
   • 設定示例：DATABASE_URL = "postgresql+asyncpg://user:pwd@host/db?maxsize=20"

5. 錯誤的例外處理

   • 在 async 函式中捕獲例外時，不要忘記 await 內部可能拋出的異常。
   • 示例：

     try:
         resp = await client.get(url)
         resp.raise_for_status()
     except httpx.HTTPError as exc:
         raise HTTPException(status_code=502, detail=str(exc))
     

實際應用場景

場景	為何選擇 async	相關程式碼片段
即時聊天服務（WebSocket）	大量長連線需要同時讀寫，async 能有效管理 I/O	from fastapi import WebSocket + await websocket.receive_text()
資料匯入/匯出（大批量 API 呼叫）	每筆請求皆為外部 API，使用 async 可一次發起多筆請求	await asyncio.gather(*tasks)
報表產生（需要查多張資料表）	多個資料庫查詢可以同時執行，縮短總時間	await database.fetch_all(query1); await database.fetch_all(query2)
圖片縮圖服務（CPU 密集）	使用 sync 處理圖片，搭配 BackgroundTasks 或 Celery 分離	background_tasks.add_task(process_image, file)
微服務間的同步呼叫	服務間的 REST 呼叫往往是 I/O，async 可減少延遲	async with httpx.AsyncClient() as client: await client.post(...)

總結

FastAPI 之所以在 Python 生態系中脫穎而出，核心就在於 同步與非同步的彈性結合。

• 同步端點簡潔、易於撰寫，適合無 I/O 或阻塞時間極短的情況。
• 非同步端點則在等待 I/O 時釋放工作者，提高併發吞吐，特別適合呼叫外部服務、資料庫或長時間的網路請求。

在實務開發時，先分析每個 API 的 I/O 特性，再決定使用 def 或 async def，並遵循以下最佳實踐：

1. 只在需要 I/O 時使用 async，避免在 CPU 密集任務中使用。
2. 所有阻塞操作必須搬到背景工作者或使用支援 await 的庫。
3. 正確管理資源的連線與關閉，確保在啟動/關閉事件中使用 await。
4. 針對高併發情境調整資料庫/Redis 連線池大小。

掌握了同步與非同步的差異與適用情境，你就能在 FastAPI 上建立既 易維護 又 高效能 的 API，為未來的微服務與雲端應用奠定堅實基礎。祝開發順利！