FastAPI集成WebSocket实时推送教程 Python Web实时数据更新方法-js教程-PHP中文网

FastAPI WebSocket 实时推送需严格管理连接生命周期、并发安全和异常退出：必须显式调用 await websocket.accept()，禁止混用 receive_* 方法，广播需用 asyncio.Lock 和 .copy() 保障线程安全，并捕获 WebSocketDisconnect 显式清理连接。

fastapi集成websocket实时推送教程 python web实时数据更新方法

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 包裹整个通信块：

必须在 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() 就会出错。

推荐统一用 receive_json() + Pydantic 模型做校验（类型安全）
若接收二进制数据（如图片上传），只能用 receive_bytes()，且后续必须用 send_bytes()
避免在一个连接里交替调用 receive_text() 和 receive_json()

多客户端广播必须用线程/协程安全的数据结构

想实现聊天室或行情推送，自然会想到用全局 list/set 存 WebSocket 对象。但 FastAPI 的 WebSocket 实例不是线程安全的，且多个协程同时读写共享集合（如 set）会导致 RuntimeError: Set changed size during iteration。

python学习

Python 编程学习助手 skill，面向零基础用户。当用户表达以下意图时触发： - 学习 Python 基础概念（变量、数据类型、循环、函数、类等） - 请求 Python 练习题或测验 - 请求审查或调试 Python 代码 - 请求 Python 项目实战指导（爬虫、数据分析、Web 开发等入门项目） - 泛化表述如"教我 Python""我想学编程""Python 怎么写""帮我学 Python" 此 skill

下载

官方推荐方案是用 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。必须显式清理：