WebSocket连接被拒绝或立即断开,根本原因在于握手失败而非心跳缺失;需确保使用@app.websocket正确声明路由、客户端发起标准ws/wss握手、服务端返回101状态码及Upgrade/Connection头,并排查Nginx代理配置是否透传升级头。

WebSocket连接为什么总是被拒绝或立即断开?
FastAPI的WebSocket对象本身不处理HTTP升级逻辑,必须用@app.websocket装饰器显式声明路由,且路径不能和普通@app.get等冲突。常见错误是误写成@app.get("/ws")再手动调用websocket.accept()——这会直接抛出RuntimeError: Not connected。
正确做法是:确保路由只被@app.websocket接管,且客户端发起的是标准WebSocket握手(ws://或wss://协议)。
- 检查浏览器控制台是否报
ERR_CONNECTION_REFUSED:确认Uvicorn服务已启动并监听正确端口(默认http://127.0.0.1:8000,但WebSocket需用ws://127.0.0.1:8000/ws) - 避免在
websocket.disconnect()后继续调用websocket.send_text(),会触发WebSocketDisconnect异常 - 若用Nginx反向代理,需显式透传
Upgrade和Connection头:location /ws {<br> proxy_pass http://backend;<br> proxy_http_version 1.1;<br> proxy_set_header Upgrade $http_upgrade;<br> proxy_set_header Connection "upgrade";<br>}
如何安全地广播消息给所有活跃连接?
FastAPI不内置连接池,需要手动维护set[WebSocket]。关键点在于:连接建立时add、断开时remove,且广播操作必须用asyncio.create_task()或async for逐个发送——不能直接await websocket.send_text()在循环里,否则一个连接卡住会阻塞全部。
示例维护方式:
创建 Flask + Vue.js 全栈项目,采用单端口架构、uv 依赖管理、结构化日志和生产级前端设计。当用户要求以下操作时触发:创建新的 Flask+Vue 项目、搭建 Web 应用、构建全栈 Python+Vue 应用、初始化 Flask Web 项目、设置单端口 Flask 前后端项目。触发词:新建项目、创建项目、搭建项目、Flask项目、Vue项目。
from fastapi import WebSocket, WebSocketDisconnect<br>from typing import Set<br><br>active_connections: Set[WebSocket] = set()<br><br>@app.websocket("/ws")<br>async def websocket_endpoint(websocket: WebSocket):<br> await websocket.accept()<br> active_connections.add(websocket)<br> try:<br> while True:<br> data = await websocket.receive_text()<br> # 广播给其他人(不含自己)<br> for conn in active_connections:<br> if conn != websocket:<br> await conn.send_text(f"User says: {data}")<br> except WebSocketDisconnect:<br> active_connections.discard(websocket)
- 用
set而非list:避免重复添加/删除时的O(n)开销 - 务必用
discard()而非remove():防止断连时连接已被移除导致KeyError - 生产环境建议加超时控制(如
await asyncio.wait_for(websocket.receive_text(), timeout=30)),防恶意长连接
为什么接收二进制数据时报TypeError: Object of type bytes is not JSON serializable?
这个错误通常出现在你把await websocket.receive_bytes()的结果直接传给json.dumps()或FastAPI的response_model。WebSocket的receive_bytes()返回原生bytes,不是JSON字符串。
- 如果客户端发的是JSON格式的二进制(比如压缩后的JSON),需先解码:
json.loads(received_bytes.decode("utf-8")) - 如果发的是纯二进制(如图片base64片段),应直接处理
bytes,不要尝试JSON序列化 - 发送时也注意:
send_text()只接受str,send_bytes()才接受bytes;混用会立刻报错 - 调试技巧:打印
type(data)和len(data),确认收发类型一致
如何让WebSocket支持查询参数或认证?
WebSocket握手请求携带的URL参数(如ws://localhost:8000/ws?token=abc123)可通过websocket.query_params获取,但它在accept()前就可用;而认证逻辑必须放在accept()之前,否则未授权连接也会被建立。
- 校验失败时,直接
raise WebSocketException(code=4001, reason="Invalid token")(code需为4000–4999),客户端会收到对应CloseEvent - JWT token验证建议用同步函数(如
jwt.decode(..., options={"verify_signature": False})快速校验结构),避免在accept()前用await阻塞握手 - Cookie认证也可读
websocket.cookies,但注意浏览器跨域时需设置credentials: true且后端响应带Access-Control-Allow-Credentials: true
复杂状态管理(如用户绑定、房间隔离)往往需要额外存储层,仅靠内存set无法支撑多进程部署——这时候该考虑Redis的Pub/Sub或专用WebSocket网关了。
大量免费API接口:立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游,为数据API应用能力赋能!










