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 Pydantic V2 資料驗證完整教學:從 BaseModel 到自訂驗證器實戰指南
相關文章
你可能也喜歡
探索其他領域的精選好文