Python FastAPI WebSocket 即時通訊開發教學：從零打造即時聊天室與通知系統

去年我在做一個內部監控儀表板的時候，需要讓數據即時更新。一開始用 polling（每秒打一次 API），結果伺服器被打爆了。後來改用 WebSocket，伺服器負載降了 80%，畫面更新也變成即時。

這篇教學會帶你用 Python FastAPI 從零開始建立 WebSocket 即時通訊應用，包含多人聊天室、廣播系統和即時通知三個實戰專案。

WebSocket 是什麼？跟 HTTP 有什麼不同？

HTTP 是「你問我答」的模式。WebSocket 則是建立一條「持久連線」，雙方都可以隨時發送訊息。WebSocket 適合即時聊天、股票報價、線上遊戲、協作編輯、即時通知等場景。

為什麼用 FastAPI 做 WebSocket？

FastAPI 原生支援 WebSocket，語法非常直覺，效能也更好（底層用 ASGI）。如果你還沒用過 FastAPI，建議先看 FastAPI REST API 入門教學。

環境建置與安裝

python -m venv venv
source venv/bin/activate
pip install fastapi uvicorn websockets

建議使用 Python 3.10 以上。如果對虛擬環境不熟悉，參考 Python 虛擬環境完全指南。

第一個 WebSocket 端點

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"收到: {data}")

實作多人聊天室

核心概念是維護一個連線清單，當有人發訊息時轉發給所有人。透過 ConnectionManager 類別管理 WebSocket 連線，每當有新使用者連線就加入清單，斷線時移除，訊息透過 broadcast 方法發送給所有在線的人。

廣播系統與連線管理

實際應用中你可能需要「房間」的概念——只有同一個房間的人才能看到彼此的訊息。可以用 defaultdict(list) 來管理不同房間的 WebSocket 連線。

即時通知系統實作

除了聊天室，WebSocket 另一個常見應用是即時通知。當後端有新事件發生（如訂單狀態更新），可以主動推送通知給前端。使用 asyncio.Queue 作為通知佇列是個好方法。

錯誤處理與斷線重連

在真實環境中，網路不穩定是常態。你的 WebSocket 應用需要用 try/except 優雅地處理 WebSocketDisconnect，並搭配 asyncio.wait_for 設定超時檢測。前端也需要實作自動重連邏輯。

部署到生產環境的注意事項

• 使用多 Worker：gunicorn -w 4 -k uvicorn.workers.UvicornWorker
• Nginx 反向代理需加 WebSocket header（Upgrade、Connection）
• 注意系統 ulimit 設定（每個 WebSocket 佔一個 file descriptor）
• 多伺服器用 Redis Pub/Sub 同步訊息

對 Python 非同步程式設計有疑問的話，建議先把 asyncio 基礎弄懂。

總結與下一步

FastAPI 的 WebSocket 支援真的很優雅。接下來你可以挑戰加入 JWT 驗證、訊息持久化，或搭配 Pandas 資料分析做即時數據儀表板。

延伸閱讀：

• Python FastAPI REST API 入門教學
• Python asyncio 非同步程式設計教學
• Python Pydantic V2 資料驗證教學