Chapter 08 · WebSocket：实时对话的桥梁
📚 学完本章，你将能够：
- 解释 HTTP 和 WebSocket 的本质区别（一次性对话 vs 持续通话）
- 用 FastAPI 建立 WebSocket 连接，收发消息
- 管理多个连接，实现广播（给所有人发消息）
- 在 WebSocket 中做用户认证
- 理解为什么 WebSocket 需要心跳保活
💡 场景引入
你的图书网站新增了"实时通知"功能：当有新书上架时，所有在线用户立即收到弹窗提醒。
用普通的 HTTP 怎么做？前端只能每隔几秒问一次服务器"有新消息吗？"——这叫轮询，既浪费带宽又延迟高。WebSocket 让服务器能主动推消息给客户端，像打电话而不是寄信。
8.1 HTTP vs WebSocket：寄信 vs 打电话
| 特性 | HTTP | WebSocket |
|---|---|---|
| 连接方式 | 每次请求新建连接（或复用） | 一次握手，长期保持 |
| 通信方向 | 客户端主动问，服务器答 | 双方随时都能发消息 |
| 头部开销 | 每次请求都带大量头部 | 握手后数据帧很小 |
| 适用场景 | 获取数据、提交表单 | 聊天、通知、股票行情、游戏 |
8.2 第一个 WebSocket 接口
WebSocket 基础
from fastapi import WebSocket, WebSocketDisconnect
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept() # 接受连接
try:
while True:
# 等待客户端发消息
data = await websocket.receive_text()
# 给客户端回消息
await websocket.send_text(f"你说了：{data}")
except WebSocketDisconnect:
print("客户端断开连接")和 HTTP 路由不同，WebSocket 路由里的 while True 会让函数一直运行，直到客户端断开。这是一个长连接。
8.3 管理多个连接：聊天室
单个连接没意思。真正的需求是：服务器同时和很多人保持通话，某个人说话时，其他人都能听到。
ConnectionManager
class ConnectionManager:
def __init__(self):
self.active_connections: list[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/chat")
async def chat(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
# 把消息发给所有人（包括自己）
await manager.broadcast(f"某人: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
并发发送的隐患
上面的 broadcast 用 for 循环逐个 await，如果连接很多，后面的用户会收到明显延迟。生产环境应该用 asyncio.gather 并发发送。
8.4 WebSocket 中的认证
浏览器在建立 WebSocket 连接时，不能像 HTTP 那样方便地自定义 Header。标准做法是通过 URL 查询参数传递 Token：
WS 认证
@app.websocket("/ws")
async def websocket_auth(websocket: WebSocket, token: str | None = None):
if not token:
await websocket.close(code=4001, reason="Missing token")
return
payload = verify_token(token)
if not payload:
await websocket.close(code=4001, reason="Invalid token")
return
await websocket.accept()
# 现在你知道这个连接是谁了
user_id = payload["sub"]
...前端连接时：wss://api.example.com/ws?token=eyJhbG...
8.5 心跳保活
Nginx 和防火墙会在连接空闲一段时间后自动切断它。WebSocket 需要心跳机制：客户端和服务端定期互发小消息，证明"我还活着"。
心跳机制
@app.websocket("/ws")
async def websocket_heartbeat(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
if data == "ping":
await websocket.send_text("pong")
continue
# 处理正常业务消息...
except WebSocketDisconnect:
...🎯 知识检测
问题 1
WebSocket 相比 HTTP 轮询的主要优势是什么？
解析：HTTP 轮询是客户端不断问"有新消息吗？"，大部分请求都是浪费。WebSocket 建立长连接后，服务器有新消息立即推送，既省带宽又低延迟。
问题 2
为什么 WebSocket 认证通常通过 URL 参数传递 Token，而不是 Header？
解析：浏览器的原生 WebSocket API（
new WebSocket(url)）不支持设置自定义 Header。虽然可以在握手阶段用 Cookie，但跨域场景受限。URL 查询参数是最通用的方案。
问题 3
心跳机制（ping/pong）的主要目的是？
解析：Nginx 默认 60 秒关闭空闲连接，防火墙也有类似策略。心跳让连接保持"活跃"状态，避免被中间设备静默断开。
本章小结
✓
HTTP 是问答，WebSocket 是对话
HTTP 一问一答就结束，WebSocket 建立后双方随时说话。
✓
ConnectionManager 管理多个连接
维护连接列表，实现广播、私聊、分组等功能。
✓
WebSocket 认证用 URL 参数
浏览器限制导致 Header 不方便，Token 放查询参数是标准做法。
✓
心跳保活防止被切断
定期 ping/pong，让 Nginx 和防火墙知道连接还在用。