FastAPI WebSockets 入门指南
在 FastAPI 中构建 WebSocket 实时功能：第一个 WebSocket 端点、连接生命周期、多连接管理与房间广播、连接鉴权，完整实战教程。
直接回答：FastAPI 不只是 REST 框架——它内置 WebSocket 支持，依赖注入、连接管理、消息广播都用同一套优雅 API。聊天室、实时通知、协同更新这类功能，几十行代码即可落地。
快速开始
pip install "fastapi[standard]" uvicorn
第一个 WebSocket 端点
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
@app.websocket("/ws/echo")
async def echo(ws: WebSocket):
await ws.accept()
try:
while True:
msg = await ws.receive_text()
await ws.send_text(f"你说: {msg}")
except WebSocketDisconnect:
pass
前端连接：
const ws = new WebSocket("ws://localhost:8000/ws/echo");
ws.onmessage = (e) => console.log(e.data);
ws.onopen = () => ws.send("你好");
连接生命周期
一次 WebSocket 连接的四个关键时刻：
- 连接到达：accept() 之前调用 close() 通常会拒绝握手并返回 HTTP 403；连接建立后才有 WebSocket 关闭码；
- 持续通信：
receive_text/bytes/json与send_*双向收发； - 断开：客户端断开时
receive抛WebSocketDisconnect，服务端主动断开调close()； - 清理：从连接管理器移除自己，释放资源，并记录关闭原因、连接持续时间和当前连接数，便于区分异常断开与正常退出。
永远用 try/except 包住接收循环——忘记处理断开异常是 WebSocket 服务日志里最常见的噪音。
多连接管理与房间广播
from collections import defaultdict
class ConnectionManager:
def __init__(self):
self.rooms: dict[str, list[WebSocket]] = defaultdict(list)
async def join(self, room: str, ws: WebSocket):
await ws.accept()
self.rooms[room].append(ws)
def leave(self, room: str, ws: WebSocket):
if ws in self.rooms[room]:
self.rooms[room].remove(ws)
async def broadcast(self, room: str, msg: str):
for ws in list(self.rooms[room]):
try:
await ws.send_text(msg)
except (WebSocketDisconnect, OSError):
self.leave(room, ws)
manager = ConnectionManager()
@app.websocket("/ws/chat/{room}")
async def chat(ws: WebSocket, room: str):
await manager.join(room, ws)
try:
while True:
msg = await ws.receive_text()
await manager.broadcast(room, msg)
except WebSocketDisconnect:
pass
finally:
manager.leave(room, ws)
注意：这个内存管理器只在单进程内有效。多副本部署时广播要经 Redis Pub/Sub 之类的通道中转，否则消息到不了其他机器上的连接。
连接鉴权
WebSocket 握手是普通 HTTP 请求，优先使用安全会话或短期一次性票据；URL 中的长期 token 会泄漏到日志。还需校验 Origin、房间访问权和消息大小：
@app.websocket("/ws/me")
async def me(ws: WebSocket, token: str | None = None):
user = verify_token(token)
if user is None:
await ws.close(code=1008)
return
await ws.accept()
...
依赖注入（Depends）在 WebSocket 路由里同样可用，但依赖须接受 WebSocket/HTTPConnection，不能假定所有基于 Request 的 HTTP 依赖可直接复用。
常见问题（FAQ）
Q：FastAPI WebSocket 能扛多少连接？
A：没有固定连接上限，需按消息频率、连接状态、内存与文件描述符限制压测。更大规模要上 Redis 中转 + 多副本 + 专门的连接层调优，或评估专用实时基础设施。
Q：和 Socket.IO 比呢？
A：Socket.IO 自带自动重连、降级轮询、房间语义，生态独立；FastAPI WebSocket 更轻、与 FastAPI 依赖体系一体。需要"断线自动续"的复杂客户端场景，Socket.IO（python-socketio）更省事。
Q：怎么给客户端推送服务端事件（非响应式）？
A：把连接存进管理器后，同一进程和事件循环中的异步任务可拿到连接并主动 send；跨线程或跨进程必须安全投递到连接所属循环——比如订单状态变更时推给对应用户。
官方参考
本文基于官方文档整理，未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全；局部片段不等同于完整生产应用。
