FastAPI: WebSocket — 实时双向通信
最后更新：2026-08-26
HTTP 像寄信——一去一回；WebSocket 像打电话——双方随时可以说话，不用挂断再拨。
1. 你将学到
- WebSocket 基础：
@app.websocket装饰器与生命周期（connect/receive/disconnect） - 连接管理器模式：
ConnectionManager类设计与广播机制 - 认证 WebSocket：在握手阶段验证 JWT Token
- 与 HTTP 端点协作：价格变更触发 WebSocket 推送
- Alice 场景：实时价格流——当百万级商品中某价格变动时，订阅用户即刻收到推送
2. Alice 的真实故事
(1) 痛点：价格变动只能轮询获取
Bob 前端每 5 秒轮询 PriceTracker 的价格接口检查是否有变动，但百万级商品中 99% 的价格在任意 5 秒内没变，轮询浪费了 99% 的请求。更糟的是，价格变动后最长需要等 5 秒才能显示，客户投诉"价格不够实时"。
(2) WebSocket 的解法
WebSocket 建立持久双向连接，服务端价格变动时主动推送给前端，无需轮询——0 浪费、0 延迟。
PYTHON
@app.websocket("/ws/prices")
async def price_websocket(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_json({"price_update": data})
(3) 收益
轮询请求从每秒 200 次降到 0，价格变动延迟从 5 秒降到 50ms，Bob 前端不再浪费 API 配额，客户满意度显著提升。
3. WebSocket 基础
(1) 生命周期状态
stateDiagram-v2
[*] --> CONNECTING: Client initiates
CONNECTING --> CONNECTED: accept()
CONNECTED --> RECEIVING: receive()
RECEIVING --> CONNECTED: send()
CONNECTED --> CLOSING: close() / disconnect
CLOSING --> CLOSED: Connection closed
CLOSED --> [*]
▶ 示例：最简 WebSocket 端点
PYTHON
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws/echo")
async def websocket_echo(websocket: WebSocket):
await websocket.accept() # Accept connection
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Echo: {data}")
except Exception:
await websocket.close()
输出：
TEXT
📖 仅展示
# 函数定义成功
▶ 示例：带路径参数的 WebSocket
PYTHON
@app.websocket("/ws/products/{product_id}/prices")
async def product_price_stream(
websocket: WebSocket,
product_id: int,
):
await websocket.accept()
try:
while True:
data = await websocket.receive_json()
# Echo back with product context
await websocket.send_json({
"product_id": product_id,
"price": data.get("price"),
"currency": data.get("currency", "USD"),
})
except Exception:
await websocket.close()
输出：
TEXT
📖 仅展示
# 函数定义成功
4. 连接管理器模式
(1) ConnectionManager 设计
▶ 示例：ConnectionManager 实现
PYTHON
from fastapi import FastAPI, WebSocket
from typing import Dict, List
import json
class ConnectionManager:
def __init__(self):
# Map: product_id -> list of active connections
self.active_connections: Dict[int, List[WebSocket]] = {}
async def connect(self, websocket: WebSocket, product_id: int):
await websocket.accept()
if product_id not in self.active_connections:
self.active_connections[product_id] = []
self.active_connections[product_id].append(websocket)
def disconnect(self, websocket: WebSocket, product_id: int):
if product_id in self.active_connections:
self.active_connections[product_id].remove(websocket)
if not self.active_connections[product_id]:
del self.active_connections[product_id]
async def broadcast_to_product(self, product_id: int, message: dict):
if product_id in self.active_connections:
dead_connections = []
for connection in self.active_connections[product_id]:
try:
await connection.send_json(message)
except Exception:
dead_connections.append(connection)
# Clean up dead connections
for conn in dead_connections:
self.disconnect(conn, product_id)
async def broadcast_all(self, message: dict):
for product_id in list(self.active_connections.keys()):
await self.broadcast_to_product(product_id, message)
manager = ConnectionManager()
输出：
TEXT
📖 仅展示
# 函数定义成功
(2) WebSocket 推送架构
flowchart TD
Update[Price Update via HTTP] --> Handler[API Handler]
Handler --> Manager[ConnectionManager]
Manager --> WS1[WebSocket Client 1]
Manager --> WS2[WebSocket Client 2]
Manager --> WSN[WebSocket Client N]
subgraph Subscribers
WS1
WS2
WSN
end
▶ 示例：WebSocket 端点使用 ConnectionManager
PYTHON
app = FastAPI()
@app.websocket("/ws/products/{product_id}/prices")
async def price_websocket(websocket: WebSocket, product_id: int):
await manager.connect(websocket, product_id)
try:
while True:
# Keep connection alive, receive any client messages
data = await websocket.receive_text()
except Exception:
manager.disconnect(websocket, product_id)
输出：
TEXT
📖 仅展示
# 函数定义成功
5. 认证 WebSocket
(1) 握手阶段验证 JWT
WebSocket 没有标准 Header 机制，Token 通过查询参数传递。
▶ 示例：WebSocket JWT 认证
PYTHON
from fastapi import WebSocket, Query, HTTPException
from jose import jwt, JWTError
from app.core.config import settings
async def verify_ws_token(token: str) -> dict:
try:
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
return payload
except JWTError:
raise ValueError("Invalid token")
@app.websocket("/ws/prices")
async def authenticated_price_ws(
websocket: WebSocket,
token: str = Query(..., description="JWT access token"),
):
# Verify token before accepting connection
try:
user = await verify_ws_token(token)
except ValueError:
await websocket.close(code=4001, reason="Authentication failed")
return
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await websocket.send_json({"user": user.get("sub"), "data": data})
except Exception:
pass
输出：
TEXT
📖 仅展示
# 函数定义成功
| 认证方式 | 实现 | 优点 | 缺点 |
|---|---|---|---|
| 查询参数 | ?token=xxx |
简单 | Token 出现在 URL 日志 |
| 首条消息 | 连接后先发 Token | 不暴露在 URL | 多一次往返 |
| Sec-WebSocket-Protocol | 子协议传 Token | 不暴露在 URL | 非标准用法 |
6. HTTP 与 WebSocket 协作
(1) 价格变更触发推送
▶ 示例：HTTP 端点触发 WebSocket 广播
PYTHON
from pydantic import BaseModel, Field
from datetime import datetime
class PriceUpdate(BaseModel):
product_id: int = Field(gt=0)
price: float = Field(gt=0, description="New price in USD")
currency: str = Field(default="USD")
source: str = Field(max_length=100)
app = FastAPI()
manager = ConnectionManager()
@app.post("/api/v1/prices", response_model=PriceResponse)
async def create_price(
price: PriceCreate,
db: AsyncSession = Depends(get_db),
user=Depends(get_current_user),
):
service = PriceService(db)
result = await service.create_price(price)
# Trigger WebSocket broadcast after successful price creation
await manager.broadcast_to_product(
product_id=price.product_id,
message={
"event": "price_update",
"product_id": price.product_id,
"new_price": price.price,
"currency": price.currency,
"source": price.source,
"timestamp": datetime.utcnow().isoformat(),
},
)
return result
输出：
TEXT
📖 仅展示
# 函数定义成功
▶ 示例：全局价格流（订阅所有变动）
PYTHON
@app.websocket("/ws/prices/stream")
async def global_price_stream(websocket: WebSocket):
await websocket.accept()
# Add to a global subscriber list
manager.global_connections.append(websocket)
try:
while True:
# Receive heartbeat/ping from client
data = await websocket.receive_text()
if data == "ping":
await websocket.send_text("pong")
except Exception:
manager.global_connections.remove(websocket)
输出：
TEXT
📖 仅展示
# 函数定义成功
❓ 常见问题
Q WebSocket 和 SSE（Server-Sent Events）怎么选？
A 需要双向通信用 WebSocket（如聊天、实时协作），只需服务端推送用 SSE（如日志流、通知）。SSE 更简单、自动重连。
Q WebSocket 连接数有上限吗？
A 单机受限于文件描述符（通常 65535）。生产环境用 Nginx 做负载均衡，多实例共享连接状态（Redis Pub/Sub）。
Q WebSocket 断线怎么处理？
A 客户端实现自动重连（指数退避）。服务端用 ping/pong 心跳检测死连接，ConnectionManager 自动清理。
Q WebSocket 能用 Pydantic 校验消息吗？
A 可以。接收消息后用
MessageModel.model_validate(data) 校验，校验失败发错误消息给客户端。Q 多个 Worker 进程如何广播？
A 单进程内存的 ConnectionManager 不够。生产环境用 Redis Pub/Sub：价格变更发布到 Redis Channel，每个 Worker 订阅并推送给本地连接。
Q WebSocket 的 CORS 怎么处理？
A 浏览器的 WebSocket 也受同源策略限制。FastAPI 的 CORSMiddleware 自动处理 WebSocket 的跨域握手。
📖 小节
- WebSocket 提供持久双向连接，替代轮询实现真正的实时推送
- ConnectionManager 管理连接池，支持按商品 ID 分组广播
- JWT 认证通过查询参数在握手阶段验证，失败直接关闭连接（code 4001）
- HTTP 端点创建价格后调用
manager.broadcast_to_product()触发推送 - 生产环境需 Redis Pub/Sub 实现跨进程广播，单进程内存不适合多 Worker
📝 作业
- 基础题（难度⭐）：创建一个 WebSocket echo 端点，客户端发送文本，服务端原样返回。用浏览器开发者工具或 wscat 测试。提示：
@app.websocket("/ws/echo")+accept()+receive_text() - 进阶题（难度⭐⭐）：实现 ConnectionManager，支持多个客户端订阅同一商品的价格变动，当新价格创建时广播给所有订阅该商品的 WebSocket 客户端。提示：
Dict[int, List[WebSocket]]+broadcast_to_product() - 挑战题（难度⭐⭐⭐）：为 WebSocket 添加 JWT 认证（查询参数传 Token），并在 HTTP 价格创建端点中触发 WebSocket 广播，实现完整的"HTTP 写入 → WebSocket 推送"链路。提示：
token: str = Query(...)+verify_ws_token()+ HTTP 端点调用manager.broadcast_to_product()
---|
