FastAPI WebSocket 实时推送需严格管理连接生命周期、并发安全和异常退出：必须显式调用 await websocket.accept()，禁止混用 receive_* 方法，广播需用 asyncio.Lock 和 .copy() 保障线程安全，并捕获 WebSocketDisconnect 显式清理连接。
FastAPI 集成 WebSocket 实现实时推送，**不是“配个装饰器就能用”——连接生命周期管理、并发安全、异常退出处理这三处不写对，服务跑两天必掉连接或内存泄漏**。
WebSocket 连接必须显式 accept()，否则客户端会卡在 pending 状态
FastAPI 的 websocket.accept() 不是可选步骤，而是协议握手的强制环节。跳过它，浏览器 WebSocket 实例状态永远停在 0（CONNECTING），onopen 不触发，send() 直接抛 InvalidStateError。
常见错误写法：
@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket):
# ❌ 忘了 accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"echo: {data}")
正确做法始终把 accept() 放在循环外第一行，并建议加 try/except 包裹整个通信块：
立即学习“Python免费学习笔记（深入）”；
- 必须在
while循环前调用await websocket.accept() - 如果需要鉴权，应在
accept()前完成（如校验Query参数或 header） -
accept()后可立即发欢迎消息，但不要在它之前调用send_*任何方法
receive_text() / receive_bytes() / receive_json() 不能混用且必须配对
同一个 WebSocket 连接中，连续调用不同类型的 receive_* 方法会导致 WebSocketDisconnect 或解析失败。FastAPI 底层依赖 websockets 库，它按帧类型严格区分文本/二进制/JSON，不自动转换。
例如：客户端发 JSON 字符串 {"msg":"hi"}，服务端若用 receive_text() 拿到的是原始字符串，需手动 json.loads()；若直接用 receive_json()，框架会帮你解析，但下一次再调 receive_text() 就会出错。
仅使用 Python 将 Claw Code 虚拟化框架与 OpenClaw 集成，提供 184 种工具和 200 多条命令，用于安全分析和代码质量...
- 推荐统一用
receive_json()+ Pydantic 模型做校验（类型安全） - 若接收二进制数据（如图片上传），只能用
receive_bytes()，且后续必须用send_bytes() - 避免在一个连接里交替调用
receive_text()和receive_json()
多客户端广播必须用线程/协程安全的数据结构
想实现聊天室或行情推送，自然会想到用全局 list/set 存 WebSocket 对象。但 FastAPI 的 WebSocket 实例不是线程安全的，且多个协程同时读写共享集合（如 set）会导致 RuntimeError: Set changed size during iteration。
官方推荐方案是用 asyncio.Lock + defaultdict(set)，但更关键的是：遍历时必须用 .copy() 防止迭代中集合被修改：
class ConnectionManager:
def __init__(self):
self.active_connections: dict[str, set[WebSocket]] = defaultdict(set)
self.lock = asyncio.Lock()
<pre class="brush:php;toolbar:false;">async def broadcast(self, room_id: str, message: str):
async with self.lock:
# ✅ copy() 避免 RuntimeError
for connection in self.active_connections[room_id].copy():
try:
await connection.send_text(message)
except WebSocketDisconnect:
await self.disconnect(room_id, connection)
- 不用
list或普通set存连接，它们不支持异步并发安全操作 - 每次广播前必须
.copy()，否则disconnect()删除连接时会崩掉迭代 - 发送前加
try/except WebSocketDisconnect，否则一个客户端断开会导致整轮广播中断
客户端断连后，服务端资源不会自动释放
WebSocket 断开（网络闪断、页面关闭、手机切后台）时，websocket.receive_* 会抛 WebSocketDisconnect 异常，但 FastAPI 不会自动从你的连接池里删掉这个 WebSocket 实例——它只是个已失效的 Python 对象，仍占着内存和事件循环引用。
后果是：连接数虚高、广播越来越慢、最终 OOM。必须显式清理：
- 所有
while True接收循环都得包在try/except WebSocketDisconnect里 - 捕获异常后，立刻调用你的连接管理器的
disconnect()方法删实例 - 别依赖
__del__或弱引用——它们不可靠，且无法保证执行时机
最易忽略的一点：广播时对每个连接调用 send_text() 前，应检查 connection.client_state == "connected"（虽然不绝对可靠，但比不查强）。真正的健壮性靠重试+心跳+超时，但那是另一层复杂度了。
