Web Day 28 WebSocket 即時更新
執行需求：CPU 可跑。前幾天的互動都是「前端打伺服器、伺服器回資料」這個單向模式。今天進入 WebSocket：「伺服器主動推送資料給前端」。這個能力讓「預約管理系統」的後台能在新預約進來時即時更新、讓聊天應用即時顯示訊息、讓儀表板即時反映數字。我們會建立完整的 FastAPI WebSocket 端點、Next.js 客戶端、連線管理與心跳機制，並用 pytest 驗證流程。
引言
Web Day 24 我們示範了 HTMX 的輪詢模式：前端每 N 秒打一次伺服器拿資料。這個模式簡單但有兩個問題：使用者看到「資料閃一下」的視覺干擾、伺服器要處理大量無意義的請求（大部分時候沒新資料）。WebSocket 解決了這兩個問題：建立一條持久連線、伺服器只在「有新資料時」推給前端，沒有新資料就不傳。
這篇文章示範一個「預約管理系統」的即時通知：當有新預約建立時，所有連線中的後台使用者會即時看到提示。我們會建立 FastAPI 的 WebSocket 端點、寫一個 ConnectionManager 管理多客戶端連線、加心跳機制偵測斷線、用 Next.js 的 client component 訂閱事件、用 pytest 驗證推送行為。WebSocket 在 2025 年 7 月的主流用法相當成熟：瀏覽器原生支援、FastAPI 內建 WebSocket 物件、uvicorn 0.35 完整支援 ASGI WebSocket。我們也會談到認證（Web Day 27 的 token 怎麼用在 WebSocket）與資源管理（連線數上限、清理）。
WebSocket 與輪詢的取捨
不是所有「即時更新」都需要 WebSocket。先看對照表：
| 特性 | 輪詢（HTMX every Ns） | SSE（Server-Sent Events） | WebSocket |
|---|---|---|---|
| 方向 | 單向（前→後） | 單向（後→前） | 雙向 |
| HTTP 相容性 | 完全 | 完全（純 HTTP） | 需 HTTP Upgrade |
| 代理伺服器支援 | 完全 | 絕大部分 | 需特別設定 |
| 瀏覽器 API | fetch |
EventSource |
WebSocket |
| 伺服器主動推送 | 否 | 是 | 是 |
| 客戶端送出 | 每次新 HTTP 請求 | 否 | 是（同連線） |
| 適合場景 | 低頻更新、可接受延遲 | 純通知、無客戶端互動 | 即時互動、聊天、協作 |
今天的範例用 WebSocket，但實務上要先回答「為什麼不用輪詢？」如果只是「每 30 秒檢查有沒有新訂單」，HTMX 輪詢就夠了；如果「使用者正在操作、需要即時回饋」（協作編輯、即時通訊、遊戲），WebSocket 才是對的工具。貫穿專案「預約管理系統」會用 WebSocket 做後台的即時通知（Web Day 39），但客戶端的預約頁面仍用輪詢或按需載入。
後端：FastAPI WebSocket 端點
FastAPI 0.116 內建 WebSocket 物件，裝飾器 @app.websocket("/ws") 把 HTTP 路徑升級為 WebSocket 連線。先看最小範例：
# ws_min.py
# 最小的 WebSocket 端點：echo 伺服器（客戶端送什麼就回什麼）
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = FastAPI()
@app.get("/")
def home():
return HTMLResponse("<h1>WebSocket echo</h1>")
@app.websocket("/ws/echo")
async def ws_echo(websocket: WebSocket):
await websocket.accept() # 接受連線
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"echo: {data}")
except WebSocketDisconnect:
# 客戶端斷線
pass這個 echo 伺服器示範 WebSocket 的最小流程：accept() 同意連線、receive_text() 等訊息、send_text() 回訊息、例外 WebSocketDisconnect 代表客戶端斷線。瀏覽器可以用 new WebSocket("ws://127.0.0.1:8000/ws/echo") 連線。
真實應用需要 ConnectionManager 管理多個連線。我們做一個完整的「預約即時通知」範例：
# ws_manager.py
# ConnectionManager：管理多客戶端 WebSocket 連線
import asyncio
import json
from typing import Any
from fastapi import WebSocket
class ConnectionManager:
def __init__(self) -> None:
# 活躍連線：所有後台使用者共用同一個廣播群
self.active: list[WebSocket] = []
self._lock = asyncio.Lock()
async def connect(self, ws: WebSocket) -> None:
await ws.accept()
async with self._lock:
self.active.append(ws)
async def disconnect(self, ws: WebSocket) -> None:
async with self._lock:
if ws in self.active:
self.active.remove(ws)
async def broadcast(self, message: dict[str, Any]) -> None:
text = json.dumps(message, ensure_ascii=False)
# 用 lock 避免同時修改 active 清單
async with self._lock:
dead: list[WebSocket] = []
for ws in self.active:
try:
await ws.send_text(text)
except Exception:
# 連線已斷、但還在 active 清單裡
dead.append(ws)
for ws in dead:
self.active.remove(ws)
async def send_personal(self, ws: WebSocket, message: dict[str, Any]) -> None:
await ws.send_text(json.dumps(message, ensure_ascii=False))
manager = ConnectionManager()ConnectionManager 集中處理三件事：connect() 把新連線加入清單、disconnect() 把斷線移除、broadcast() 把訊息送給所有活躍連線。asyncio.Lock 保護「修改清單」這個動作的執行緒安全（多個連線同時進來不會壞掉）。broadcast() 內部 try/except 把「送訊息失敗」的連線標記為 dead、從清單移除，避免下次廣播又送給斷線的連線。
整合進 FastAPI 應用：
# main_ws.py
# FastAPI 應用：預約即時通知（WebSocket）
import asyncio
import json
from datetime import datetime, timezone
from fastapi import Depends, FastAPI, HTTPException, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from main_auth import get_current_user # 沿用 Web Day 27 的認證
from ws_manager import manager
from services import create_booking_service
from main_api import BookingIn
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.post("/api/bookings")
async def create_booking(payload: BookingIn):
booking = create_booking_service(payload)
# 廣播給所有 WebSocket 訂閱者
await manager.broadcast({
"type": "booking.created",
"data": booking.model_dump(),
"ts": datetime.now(timezone.utc).isoformat(),
})
return booking
@app.websocket("/ws/notifications")
async def notifications(websocket: WebSocket):
# 從 query string 拿 access token（WebSocket 不能用 cookie 預設機制）
token = websocket.query_params.get("token")
if not token:
await websocket.close(code=1008) # policy violation
return
# 驗 token（簡化版，直接解碼；正式環境用 main_auth.get_current_user）
try:
import jwt
data = jwt.decode(token, "dev-secret-change-me", algorithms=["HS256"])
except Exception:
await websocket.close(code=1008)
return
await manager.connect(websocket)
try:
# 傳送歡迎訊息
await websocket.send_text(json.dumps({
"type": "welcome",
"user": data["sub"],
"ts": datetime.now(timezone.utc).isoformat(),
}, ensure_ascii=False))
while True:
# 客戶端可以送訊息（這裡只處理 heartbeat）
raw = await websocket.receive_text()
try:
msg = json.loads(raw)
except json.JSONDecodeError:
continue
if msg.get("type") == "ping":
await websocket.send_text(json.dumps({
"type": "pong",
"ts": datetime.now(timezone.utc).isoformat(),
}, ensure_ascii=False))
except WebSocketDisconnect:
await manager.disconnect(websocket)這個範例做了幾件事。/api/bookings POST 端點在新增預約後呼叫 manager.broadcast()，把所有 WebSocket 訂閱者都通知「有新預約」；/ws/notifications 是 WebSocket 端點，從 query string 拿 access token（瀏覽器原生 WebSocket 不像 fetch 能自動帶 cookie，所以 token 通常放 query string 或第一幀訊息）、驗證、加入 ConnectionManager。1008 是 WebSocket 的 policy violation close code，1000 是 normal closure，1011 是 server error。
WebSocket 與 HTTP 認證的差異值得說明：瀏覽器原生 WebSocket 物件不會自動帶 cookie，所以需要把 token 放在 query string（ws://host/ws?token=xxx）或第一幀訊息裡。query string 的 token 會留在伺服器 log，所以要特別小心 log filter（Web Day 22 處理）。第一幀訊息（subprotocol 訊息）相對安全但需要前端配合；如果可以，用 cookies（搭配 SameSite）並由瀏覽器自動帶仍是較安全的選擇，但瀏覽器對 WebSocket cookie 的處理不一致，實務上仍常見 query string 或首幀訊息。
前端：Next.js 的 WebSocket 客戶端
Next.js 15 的 WebSocket 必須在 client component 裡建立（瀏覽器原生 API）。我們做一個通用的 hook 處理連線、重連、心跳：
// src/lib/useWebSocket.ts
// 自動重連與心跳的 WebSocket hook
"use client";
import { useEffect, useRef, useState } from "react";
type Handler = (msg: unknown) => void;
export function useWebSocket(url: string, onMessage: Handler) {
const [status, setStatus] = useState<"connecting" | "open" | "closed">(
"connecting",
);
const wsRef = useRef<WebSocket | null>(null);
const onMessageRef = useRef(onMessage);
onMessageRef.current = onMessage;
useEffect(() => {
let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
let pingTimer: ReturnType<typeof setInterval> | null = null;
let cancelled = false;
function connect() {
if (cancelled) return;
const ws = new WebSocket(url);
wsRef.current = ws;
setStatus("connecting");
ws.onopen = () => {
setStatus("open");
// 每 25 秒送一次 ping，避免被中介設備關閉
pingTimer = setInterval(() => {
if (ws.readyState === WebSocket.OPEN) {
ws.send(JSON.stringify({ type: "ping" }));
}
}, 25_000);
};
ws.onmessage = (event) => {
try {
const data = JSON.parse(event.data);
onMessageRef.current(data);
} catch (e) {
// 不是 JSON 就忽略
}
};
ws.onclose = () => {
setStatus("closed");
if (pingTimer) clearInterval(pingTimer);
if (!cancelled) {
// 3 秒後重連
reconnectTimer = setTimeout(connect, 3_000);
}
};
ws.onerror = () => {
ws.close();
};
}
connect();
return () => {
cancelled = true;
if (reconnectTimer) clearTimeout(reconnectTimer);
if (pingTimer) clearInterval(pingTimer);
wsRef.current?.close();
};
}, [url]);
return { status };
}這個 hook 處理三件事：連線（new WebSocket(url)）、心跳（每 25 秒送 ping）、斷線重連（3 秒後重試）。cancelled flag 確保 React unmount 時不會繼續重連。onMessageRef.current = onMessage 是 React 的標準模式：把 callback 放進 ref，讓 effect 不依賴 callback 的 identity、避免無限迴圈。
使用 hook 訂閱預約通知：
// src/app/admin/notifications.tsx
// 後台通知元件：訂閱 WebSocket，新預約即時顯示
"use client";
import { useState } from "react";
import { useWebSocket } from "@/lib/useWebSocket";
type Notification = {
id: number;
customer: string;
slot: string;
ts: string;
};
export function Notifications() {
const [items, setItems] = useState<Notification[]>([]);
const { status } = useWebSocket(
"ws://127.0.0.1:8000/ws/notifications?token=ACCESS_TOKEN",
(msg) => {
const m = msg as { type: string; data?: { customer: string; slot: string }; ts: string };
if (m.type === "booking.created" && m.data) {
setItems((prev) => [
{
id: Date.now(),
customer: m.data!.customer,
slot: m.data!.slot,
ts: m.ts,
},
...prev,
].slice(0, 10)); // 只保留最近 10 筆
}
},
);
return (
<div>
<p>連線狀態：{status}</p>
<ul>
{items.map((it) => (
<li key={it.id}>
{it.customer}｜{it.slot}｜{it.ts}
</li>
))}
</ul>
</div>
);
}這個元件用 useWebSocket 訂閱 /ws/notifications，當收到 booking.created 訊息就把新條目插到清單頂端。token 的部分簡化成 query string 帶 ACCESS_TOKEN，實務上由父元件從 httpOnly cookie 解出（或由 Server Component 注入到 client 端）。
用 pytest 驗證 WebSocket 流程
WebSocket 的測試比 HTTP 端點複雜：需要模擬客戶端連線、收訊息、驗證行為。FastAPI 的 TestClient 對 WebSocket 支援完整：
# test_ws.py
# WebSocket 端點測試
import json
import time
import jwt
import pytest
from fastapi.testclient import TestClient
from main_ws import app
client = TestClient(app)
@pytest.fixture
def access_token():
# 簽一個測試用的 access token（沿用 main_auth 的 SECRET）
return jwt.encode(
{"sub": "alice", "exp": int(time.time()) + 300, "type": "access"},
"dev-secret-change-me",
algorithm="HS256",
)
def test_websocket_receives_welcome(access_token):
with client.websocket_connect(
f"/ws/notifications?token={access_token}"
) as ws:
msg = ws.receive_text()
data = json.loads(msg)
assert data["type"] == "welcome"
assert data["user"] == "alice"
def test_websocket_ping_pong(access_token):
with client.websocket_connect(
f"/ws/notifications?token={access_token}"
) as ws:
# 跳過 welcome
ws.receive_text()
# 送 ping
ws.send_text(json.dumps({"type": "ping"}))
# 收 pong
data = json.loads(ws.receive_text())
assert data["type"] == "pong"
def test_websocket_without_token():
from starlette.websockets import WebSocket
# 沒 token 會被 close 1008
with pytest.raises(WebSocket):
with client.websocket_connect("/ws/notifications"):
pass
def test_websocket_receives_booking_broadcast(access_token):
with client.websocket_connect(
f"/ws/notifications?token={access_token}"
) as ws:
# 跳過 welcome
ws.receive_text()
# 另一個 client 建立新預約，應觸發廣播
r = client.post(
"/api/bookings",
json={"customer": "王小明", "slot": "2025-08-10T10:00"},
)
assert r.status_code == 201
# WebSocket 應收到 booking.created
msg = ws.receive_text()
data = json.loads(msg)
assert data["type"] == "booking.created"
assert data["data"]["customer"] == "王小明"這份測試做四件事：test_websocket_receives_welcome 驗證連線後收到歡迎訊息、test_websocket_ping_pong 驗證心跳機制、test_websocket_without_token 驗證沒 token 被拒、test_websocket_receives_booking_broadcast 驗證「POST 預約後 WebSocket 收到廣播」這個端到端流程。with client.websocket_connect(...) as ws 是 TestClient 的特殊 context manager，自動處理連線與清理；ws.receive_text() 與 ws.send_text() 是雙向通訊介面。
端到端測試要特別注意：建立 POST 與 WebSocket 連線要在同一個 process 內，才能共享 ConnectionManager 的 in-memory 清單。TestClient 預設是同步的，不會跑事件迴圈；FastAPI 的 WebSocket 會在背景任務裡廣播，TestClient 的同步模型可能會錯過訊息。實務上可能需要 pytest-asyncio 或等待一下時間。
常見錯誤與踩雷
第一個常見踩雷：瀏覽器連線失敗但沒錯誤訊息。ws.onerror 只會收到一個 Event 物件，沒有具體原因。常見原因：CORS 沒設（WebSocket 不受 CORS 影響，但要伺服器允許）、URL 協定錯（HTTP 網頁用 ws://、HTTPS 用 wss://）、伺服器沒啟動。對應排查：用 wscat（CLI 工具）測連線；wscat -c ws://127.0.0.1:8000/ws/notifications。
第二個常見踩雷：心跳沒做、連線被中間設備關閉。許多反向代理（nginx、CDN）會在閒置 60 秒後關閉連線。我們今天用 25 秒心跳避開這個問題。如果沒做心跳，使用者會看到「通知停了一段時間又突然恢復」。
第三個常見踩雷：廣播時遇到慢客戶端卡住整個連線池。await ws.send_text() 是阻塞的，如果某個客戶端網路慢，整個 broadcast() 迴圈會等他送完才繼續。對應策略：用 asyncio.gather() 並行送、或設 timeout 跳過慢客戶端。實務上 ConnectionManager 加個超時機制是必要的：
# 改良版 broadcast：並行送、慢客戶端就跳過
import asyncio
async def broadcast(self, message: dict[str, Any]) -> None:
text = json.dumps(message, ensure_ascii=False)
async with self._lock:
# 複製清單避免送出時被改
clients = list(self.active)
async def safe_send(ws: WebSocket) -> WebSocket | None:
try:
# 0.5 秒沒送出就當斷線
await asyncio.wait_for(ws.send_text(text), timeout=0.5)
return None
except Exception:
return ws
# 並行送所有客戶端
dead = await asyncio.gather(*(safe_send(ws) for ws in clients))
if dead:
async with self._lock:
for ws in dead:
if ws and ws in self.active:
self.active.remove(ws)asyncio.gather() 同時對所有客戶端送訊息，總時間是最慢那個（不再累加）；慢客戶端超過 0.5 秒就被視為斷線、從清單移除。Web Day 20 會用 Redis 取代 in-memory 儲存，多機部署時也能正常運作。
第四個常見踩雷：WebSocket 沒處理客戶端傳來的無效訊息。ws.receive_text() 接到非 JSON 會拋例外，整個連線就斷了。對應策略：用 try/except 把單一訊息的錯誤吃掉、繼續迴圈，不要因為一個訊息壞了整條連線。
第五個常見踩雷：忘了 WebSocketDisconnect 例外。客戶端主動關閉連線時，伺服器會收到 WebSocketDisconnect，沒接會印一堆 traceback 但連線還是會清掉（FastAPI 會處理）。對應策略：明確 except WebSocketDisconnect 並 await manager.disconnect(ws)。
效能與實務提醒
WebSocket 連線數會消耗伺服器資源（每條連線一個 task、約 10–50 KB 記憶體）。一個 FastAPI process 預設能撐幾千條同時連線，但若應用在「百萬級活躍使用者」場景，要考慮水平擴展 + Redis pub/sub（Web Day 20 與貫穿專案的後台會展開）。
另一個常見的設計陷阱：「所有資料都走 WebSocket」。WebSocket 適合「即時事件」「雙向互動」，但「頁面初次載入」應該走 HTTP（Day 25 的讀取模式）。混合兩者才是務實做法：HTTP 拿初始資料、WebSocket 接收後續更新。我們今天的範例就是這樣：GET /api/bookings 拿初始清單、WebSocket 收 booking.created 增量更新。
最後一個提醒：WebSocket 部署需要 Reverse Proxy 支援（Web Day 33 展開）。nginx 需要 proxy_set_header Upgrade $http_upgrade 與 proxy_set_header Connection "upgrade" 才會把 HTTP 升級成 WebSocket；Caddy 預設支援 WebSocket。沒有正確設定的話，連線會在 426 Upgrade Required 失敗。
小結
今天把 WebSocket 從「概念」推到「完整實作」。我們比較了輪詢、SSE、WebSocket 三種即時更新方案；建立 ConnectionManager 管理多客戶端連線；用 FastAPI 0.116 的 @app.websocket 寫通知端點；用 Next.js 15 的 client component + custom hook 訂閱事件；用 pytest 測試歡迎訊息、心跳、廣播、未授權斷線。今天最重要的是「WebSocket 不是輪詢的替代品，是另一種工具」——選對場景才有效。後天（Web Day 29）我們會進入「上線」前的準備：環境變數、設定管理、怎麼把 dev 設定換成 prod 設定而不改程式碼。
壓力測試 WebSocket
WebSocket 的連線數是系統容量規劃的關鍵指標。我們用 Python + httpx 寫一支簡單的壓力測試，模擬 100 個客戶端同時連線並維持 30 秒：
# scripts/ws_load.py
# 壓力測試：模擬多個 WebSocket 客戶端
import asyncio
import time
import jwt
import websockets
URL = "ws://127.0.0.1:8000/ws/notifications?token={token}"
NUM_CLIENTS = 100
DURATION_SECONDS = 30
async def client_loop(token: str, stats: dict) -> None:
url = URL.format(token=token)
try:
async with websockets.connect(url) as ws:
await ws.recv() # welcome
stats["connected"] += 1
end = time.time() + DURATION_SECONDS
while time.time() < end:
await ws.send('{"type":"ping"}')
await asyncio.wait_for(ws.recv(), timeout=5)
stats["completed"] += 1
except Exception as e:
stats["errors"] += 1
async def main() -> None:
token = jwt.encode(
{"sub": "load-test", "exp": int(time.time()) + 600, "type": "access"},
"dev-secret-change-me",
algorithm="HS256",
)
stats = {"connected": 0, "completed": 0, "errors": 0}
tasks = [client_loop(token, stats) for _ in range(NUM_CLIENTS)]
await asyncio.gather(*tasks, return_exceptions=True)
print(f"總客戶端：{NUM_CLIENTS}")
print(f"成功連線：{stats['connected']}")
print(f"完成測試：{stats['completed']}")
print(f"錯誤次數：{stats['errors']}")
if __name__ == "__main__":
asyncio.run(main())這支腳本展示 WebSocket 壓力測試的最小寫法：用 websockets 函式庫（不是 httpx，httpx 還沒原生支援 WebSocket 客戶端）、asyncio 並行 100 條連線、每條維持 30 秒送 ping。執行前要先 pip install websockets==13。如果你的 Web Day 43 壓力測試章節會正式展開 locust，這支腳本就是 locust 的簡化版先驅。
連線數上限與資源管理
WebSocket 連線會佔用伺服器資源，每條連線都需要一個 asyncio task、一些 buffer、與一個 socket。一個 FastAPI process 預設能撐 5000–10000 條同時連線，但這是上限不是目標。實務上我們會設兩個保護機制：每個使用者只能 N 條連線、整體連線數上限 M。超出時拒絕新連線或踢掉舊連線。
把這個邏輯加進 ConnectionManager：
# ws_limits.py
# 連線數上限控制
MAX_PER_USER = 3
MAX_TOTAL = 1000
class LimitedConnectionManager(ConnectionManager):
def __init__(self) -> None:
super().__init__()
# user_id -> 連線數
self._per_user: dict[str, int] = {}
async def connect(self, ws: WebSocket, user_id: str) -> bool:
# 總數上限檢查
if len(self.active) >= MAX_TOTAL:
await ws.close(code=1013) # try again later
return False
# 單一使用者上限檢查
if self._per_user.get(user_id, 0) >= MAX_PER_USER:
await ws.close(code=1008) # policy violation
return False
await ws.accept()
async with self._lock:
self.active.append(ws)
self._per_user[user_id] = self._per_user.get(user_id, 0) + 1
return True
async def disconnect(self, ws: WebSocket, user_id: str) -> None:
await super().disconnect(ws)
self._per_user[user_id] = max(0, self._per_user.get(user_id, 1) - 1)1013 try again later 是告訴客戶端「伺服器暫時忙、稍後再試」；1008 policy violation 是「違反政策」這裡用於「單一使用者太多連線」。把 user_id 從 token 解出、傳給 connect/disconnect，就能正確清理計數。
另一個資源管理重點是「idle connection 自動斷線」。如果使用者關閉瀏覽器但網路不穩定，伺服器可能要等 TCP timeout 才發現連線死了。我們用「伺服器端逾時」主動關閉：
# ws_idle.py
# idle connection 自動斷線
import asyncio
class IdleKiller:
def __init__(self, manager: ConnectionManager, timeout: float = 60.0) -> None:
self.manager = manager
self.timeout = timeout
self._last_seen: dict[int, float] = {}
self._task: asyncio.Task | None = None
async def _watch(self) -> None:
while True:
await asyncio.sleep(10)
now = time.time()
for ws, last in list(self._last_seen.items()):
if now - last > self.timeout:
await ws.close(code=1000)
del self._last_seen[id(ws)]
def touch(self, ws: WebSocket) -> None:
self._last_seen[id(ws)] = time.time()
def start(self) -> None:
self._task = asyncio.create_task(self._watch())每收到一個訊息就呼叫 touch(ws) 更新最後活躍時間；背景任務每 10 秒掃描、把超過 timeout 的連線關閉。60 秒是常見值，反向代理（nginx）通常也是 60 秒關閉閒置連線，配對起來剛好。Web Day 33 的反向代理設定會對應這個 timeout。
用 pytest 驗證連線上限
連線數上限是「壓力情境下才會觸發」的功能，但也要有測試保護避免日後被改壞。我們用 pytest + asyncio 模擬多個客戶端同時連線：
# test_limits.py
# 連線上限測試
import asyncio
import jwt
import pytest
import websockets
URL = "ws://127.0.0.1:8000/ws/notifications?token={token}"
SECRET = "dev-secret-change-me"
@pytest.fixture
def token():
import time
return jwt.encode(
{"sub": "alice", "exp": int(time.time()) + 600, "type": "access"},
SECRET,
algorithm="HS256",
)
@pytest.mark.asyncio
async def test_under_limit_connects():
# 預期 100 條都能連線成功（伺服器預設上限遠高於此）
tokens = [token] * 100
connections = await asyncio.gather(
*(websockets.connect(URL.format(token=t)) for t in tokens),
return_exceptions=True,
)
success = [c for c in connections if not isinstance(c, Exception)]
assert len(success) >= 50 # 寬鬆斷言，避免 CI 環境不穩定
# 關閉所有連線
for c in success:
await c.close()
@pytest.mark.asyncio
async def test_per_user_limit_rejects_excess():
# 單一使用者開 5 條連線，預期超過 3 條的會被 close
connections = []
for _ in range(5):
try:
ws = await websockets.connect(URL.format(token=token))
connections.append(ws)
except Exception:
connections.append(None)
# 至少有一些成功、一些被拒
success = [c for c in connections if c is not None]
assert 0 < len(success) < 5
for c in success:
await c.close()這份測試展示 WebSocket 並行測試的兩個模式：用 asyncio.gather 同時建立多條連線、用 return_exceptions=True 把失敗也當結果收集起來（避免一個失敗讓整個測試當掉）。pytest-asyncio 是 pytest 的 async 支援套件，要先 uv add --dev pytest-asyncio==0.24。實務上壓力測試通常不會跑在 CI（環境不穩定），而是在 staging 環境用 locust 跑。
多機部署的關鍵考量
單機 FastAPI process 的 WebSocket 數量有限，要支援更多使用者需要水平擴展（多機部署）。這帶來新問題：使用者 A 連到 A 機、新預約發生在 B 機、A 機的 ConnectionManager 沒收到通知。解法是「訊息匯流排」（message bus），最常用 Redis Pub/Sub。
Web Day 20 會展開 Redis，今天先看概念：當 POST /api/bookings 收到新預約時，不直接呼叫 manager.broadcast()，而是 redis.publish("bookings", json)；每台 FastAPI 機器的 WebSocket 端點訂閱 "bookings" channel、收到訊息後對自己的連線廣播。這樣新預約會被所有機器收到、所有連線都被通知到。
另一個關鍵：sticky session。WebSocket 連線建立後會留在同一台機器處理，nginx 通常需要 ip_hash 設定確保同一個 client 一直連到同一台機器。如果沒有 sticky session，使用者每次重新連線可能落到不同機器，訊息就漏掉。Web Day 33 的反向代理會展開 sticky session 的設定。
真實場景的設計建議
把 WebSocket 用在「對」的場景才有意義。我們整理三個常見的真實場景與建議。
第一個場景是「即時儀表板」。例如管理者打開後台看到「目前有 5 個客戶正在預約」「最近 10 分鐘新增 12 筆預約」這種數字。WebSocket 適合做這件事：伺服器每 1 秒計算數字、廣播給所有連線者；客戶端用 useState 顯示。但要注意「不要每個客戶端都計算同一份數字」——這個場景應該由伺服器算一次、廣播給 N 個客戶端，節省 CPU。
第二個場景是「協作通知」。例如 A 編輯預約、B 正在看同一個預約頁面，A 儲存後 B 看到「已被更新，請重新整理」。這種通知頻率低（每分鐘幾次）、訊息簡單（「預約 #123 已更新」）、不需要雙向互動。WebSocket 適合，但 SSE 或輪詢也合理。
第三個場景是「雙向即時互動」。例如客戶與客服的即時對話、教練與學員的即時互動。這種場景訊息量大、需要雙向（雙方都要送訊息）、延遲敏感（打字就要看到）。WebSocket 是唯一合理選擇，SSE 與輪詢都做不到。
第四個場景（不適合 WebSocket）：批次資料匯入後通知使用者「匯入完成」。這個場景只需要通知一次，用 Server-Sent Events 或 webhook 都比 WebSocket 簡單；WebSocket 要維護持久連線、處理心跳，成本不成比例。
結語
今天的重點是「即時更新的三條路」。我們比較了輪詢、SSE、WebSocket 的取捨，選擇 WebSocket 做示範。建立 ConnectionManager 管理連線、加心跳、處理廣播時的慢客戶端、用 pytest 驗證整個流程。讀完這篇你應該能回答：輪詢與 WebSocket 的差別是什麼？為什麼要心跳？ConnectionManager 為什麼需要 Lock？WebSocket 的 token 認證怎麼做？
明天，我們進入「環境變數與設定管理」。Web Day 27 已經把 SECRET 寫在程式碼裡（SECRET = "dev-secret-change-me"），這在正式環境是嚴重錯誤。明天會用 pydantic-settings 把所有設定外部化：dev、staging、prod 三套環境用不同設定、敏感資訊從環境變數讀、配置改變不需要改程式碼。這是從「本機開發」走到「正式上線」的關卡，請預留完整的時間跟著做。
延伸資源
- FastAPI WebSocket 官方文件（0.116，2025）：
https://fastapi.tiangolo.com/advanced/websockets/，accept、receive_text、send_text、WebSocketDisconnect的完整用法。 - MDN WebSocket API（2025）：
https://developer.mozilla.org/zh-TW/docs/Web/API/WebSocket，瀏覽器端的onopen/onmessage/onclose事件、close code 全清單。 - RFC 6455 WebSocket 協定（2011）：
https://datatracker.ietf.org/doc/html/rfc6455，協定層的完整定義，包含 frame 格式與升級流程。 - asyncio.gather 與 wait_for 官方文件（2025）：
https://docs.python.org/3/library/asyncio-task.html，並行任務與超時控制。 - Starlette WebSocket 測試（2025）：
https://fastapi.tiangolo.com/advanced/testing-websockets/，TestClient 的 WebSocket 測試慣例。
留言
張貼留言