Python实现实时WebSocket通信_FastAPI与客户端协同

FastAPI 实现 WebSocket 实时通信需服务端管理连接生命周期、支持广播/定向消息，客户端实现稳定连接与自动重连；原生支持无需额外框架，配合 HTML+JS 即可完*链路。

用 FastAPI 实现 WebSocket 实时通信，关键在于服务端正确管理连接生命周期、支持广播或定向消息，同时客户端能稳定连接、自动重连并处理收发逻辑。不需要额外框架，FastAPI 原生支持 WebSocket，配合现代前端（如 HTML + JS）即可跑通完整链路。

服务端：用 FastAPI 建立 WebSocket 路由与连接池

FastAPI 的 WebSocket 依赖可直接注入路由函数，每个连接是一个独立的协程。建议用全局集合（如 set）暂存活动连接，便于后续广播；注意连接断开时及时清理，避免内存泄漏。

• 使用 websocket.accept() 启动连接，不能漏掉，否则握手失败
• 用 async for message in websocket.iter_text() 或 iter_json() 持续读取消息，别用阻塞式循环
• 广播时对每个连接 await websocket.send_text(...)，需包裹 try/except 捕获 WebSocketDisconnect 和连接关闭异常
• 示例中可加一个简单计数器或房间 ID 字段，区分不同会话组

客户端：浏览器原生 WebSocket + 基础重连机制

前端无需第三方库，new WebSocket(url) 即可连接。重点是处理就绪状态、错误恢复和消息解析。默认不自动重连，需手动实现退避策略（如指数增长延迟）。

• 监听 onopen 确认连接成功，再发送初始化消息（如用户 ID 或加入房间请求）
• onmessage 中用 JSON.parse() 解析后端发来的 JSON 字符串，避免直接渲染未校验内容
• onclose 和 onerror 触发时启动重连，限制最大重试次数（如 5 次），超时后提示用户手动刷新
• 发送前检查 ws.readyState === WebSocket.OPEN，非 OPEN 状态缓存待发消息或丢弃

消息协议：轻量结构 + 类型标识

前后端约定简单 JSON 格式，避免歧义。推荐包含 type 字段区分消息语义（如 "chat"、"join"、"pong"），再带 data 载荷。服务端按 type 分发逻辑，客户端按 type 更新 UI。

• 例如客户端发：{"type": "join", "data": {"room": "general", "user": "alice"}}
• 服务端响应：{"type": "welcome", "data": {"online": 12}}
• 聊天消息：{"type": "chat", "data": {"from": "bob", "text": "Hi!", "ts": 1718234567}}
• 心跳保活可双向发 {"type": "ping"} / {"type": "pong"}，超时未响应则主动断连

部署与调试：跨域、路径与日志要点

开发时常见问题多出在 CORS 和路径配置。FastAPI 默认不限制 WebSocket 跨域，但需确认反向代理（如 Nginx）未拦截 Upgrade 请求；生产环境建议固定 WebSocket 路径（如 /ws），避免与 HTTP 路由冲突。

• 启动命令加 --reload 支持热更新，但 WebSocket 连接会在重启时断开，属正常行为
• 用 logging.info(f"Client {client_id} connected") 记录连接/断开，比 print 更易追踪
• 浏览器开发者工具的 Network → WS → Messages 面板可实时查看帧收发，验证协议是否符合预期
• 若遇 Error during WebSocket handshake: Unexpected response code: 403，检查是否漏了 websocket.accept() 或中间件拦截了 Upgrade 头