FastAPI 並發與性能優化：Async、Await 與 Event Loop 運作機制

FastAPI 最引以為傲的特性之一就是它的 效能。它宣稱自己是目前最快的 Python Web 框架之一，效能足以與 NodeJS 和 Go 並駕齊驅。

這背後的秘密在於它是如何處理 「並發 (Concurrency)」 的。

理解 Async/Await 與 Event Loop

FastAPI 是基於 ASGI (Asynchronous Server Gateway Interface) 標準所建構的。這意味著它可以處理大量併發的連線，而不會為每個連線開啟新的執行緒，從而節省大量系統資源。

核心觀點：餐廳的比喻

• 同步 (Sync/Threaded)：一個服務生點完菜後，在廚房門口「等」菜做好才送出去。如果這道菜要寫很久，這個服務生就不能服務其他客人。
• 非同步 (Async)：一個服務生點完菜轉身就去服務下一桌客人。廚房把菜做好了，服務生再回來送菜。

FastAPI 的魔法：async def vs def

這是許多開發者最困惑的部分。在 FastAPI 中，你可以定義路由為 async def 或普通的 def。FastAPI 處理這兩者的方式截然不同：

1. async def (非同步路徑)

如果你定義了 async def，FastAPI 會直接在 Event Loop (事件迴圈) 中執行它。

@app.get("/")
async def read_results():
    # 這裡必須是 non-blocking (非同步) 的代碼
    results = await some_library.fetch_data()
    return results

2. def (同步路徑)

如果你定義了普通的 def，FastAPI 不會直接執行它，而是會將它丟入一個 內部的執行緒池 (ThreadPool) 中執行。

@app.get("/")
def read_results():
    # 這裡可以放會卡住的同步代碼，FastAPI 會幫你分散到執行緒跑
    import time
    time.sleep(1)
    return {"message": "done"}

這是一個非常聰明的設計。即使你不懂 async/await，只要用普通的 def，FastAPI 會保護你不去卡死 Event Loop。

如何做好平行處理的最佳化？

如果你追求極致的效能，你應該儘量使用 async def 並搭配非同步的 Library（如 httpx 取代 requests，motor 取代 pymongo）。

但如果你必須使用某些只有「同步」介面的老舊 Library（例如傳統的 PIL 或舊版 SDK），你該怎麼辦？

做法 A：使用普通的 def (懶人首選)

如上文所述，讓 FastAPI 自動幫你丟到 ThreadPool。

做法 B：在 async def 中手動使用 run_in_executor

如果你已經身處一個 async def 函式中，但中間需要執行一小段會卡住的程式碼（例如：呼叫舊版 SDK、讀取大型檔案、或是處理圖片），你不能直接呼叫它。這時你需要將該任務「委託」給 Loop 的執行緒池。

import asyncio

# 這是同步函式，會卡住 2 秒
def sync_blocking_task(name: str, delay: float):
    import time
    time.sleep(delay)
    return f"Hello {name}, after {delay} seconds"

@app.get("/run-sync")
async def run_sync_in_async():
    # 1. 取得當前正在運行的 Event Loop
    loop = asyncio.get_running_loop()

    # 2. 委託執行
    # 第一個參數：Executor。傳入 None 表示使用預設的 ThreadPoolExecutor
    # 第二個參數：要執行的函式名稱 (注意：不要加括號，不是呼叫它)
    # 第三個以後的參數：會被當作位置參數傳給該函式
    result = await loop.run_in_executor(
        None,
        sync_blocking_task,
        "FastAPI User",
        2.0
    )

    return {"message": result}

常見應用場景 (Use Cases)

1. 處理圖像 (Pillow)：圖片裁切、壓縮是運算密集且通常是同步 IO，直接寫在 async def 會讓所有 User 一起卡住。

   from PIL import Image
   
   def process_image(path):
       with Image.open(path) as img:
           img = img.resize((100, 100))
           img.save("thumbnail.jpg")
   
   # 在路由中使用 loop.run_in_executor(None, process_image, "photo.jpg")
   

2. 呼叫舊版 SDK：許多雲端服務或資料庫的舊版 SDK 只有同步介面。

   def legacy_search(query):
       # 假設這個 SDK 內部使用 requests 且沒有 async 版本
       return client.search_data(query)
   

ThreadPool vs ProcessPool

• None (預設)：使用 ThreadPoolExecutor。適合「等待」型任務（IO 密集），如請求 API、讀取硬碟。這也是最常用的選項。
• ProcessPoolExecutor：適合真正的「運算」型任務（CPU 密集），如加解密、大矩陣運算。使用進程池可以避開 Python 的 GIL (全域解釋器鎖) 限制。

from concurrent.futures import ProcessPoolExecutor

executor = ProcessPoolExecutor()

@app.get("/heavy-math")
async def heavy_math():
    loop = asyncio.get_running_loop()
    # 傳入指定 executor，讓任務在獨立的子進程中跑
    result = await loop.run_in_executor(executor, math_func)
    return result

總結：最佳實踐清單

1. 優先使用非同步：只要有非同步版本的 Library (例如 tortoise-orm, databases, httpx)，就優先使用它們並搭配 async def。
2. 遇到同步不要怕：如果 Library 只有同步版，就用普通的 def。FastAPI 的 ThreadPool 預設效能已經非常優秀。
3. 不要在非同步中呼叫同步：這點最重要。在 async def 裡寫 requests.get() 是自廢武功的行為。
4. CPU 密集型任務：如果是大量運算，即使丟到 ThreadPool 也會受限於 Python 的 GIL。建議使用 ProcessPoolExecutor 或將任務委派給 Background Tasks 或 Celery。