17370845950

Python WebSocket实时通信核心是websockets库服务端与原生API/ websocket-client客户端，10行代码可实现单连接回显，浏览器或Python客户端均可测试，需注意跨域、连接保持、编码和广播等常见问题。

Python 中实现 WebSocket 实时通信，核心在于服务端用 websockets 库（轻量、标准兼容），客户端可用浏览器原生 API 或 Python 的 websocket-client。不需要 Flask/Django 也能跑通基础双向通信，关键在理解连接生命周期和消息收发逻辑。

搭建最小可行 WebSocket 服务端

用 websockets 库启动一个单连接回显服务只需 10 行代码：

• 安装：pip install websockets
• 新建 server.py，定义处理函数接收消息并原样发回
• 调用 serve() 绑定地址（如 localhost:8765）并运行事件循环
• 该服务默认只处理一个连接；如需多客户端广播，需自行维护连接集合

浏览器直连测试（无需额外客户端）

打开浏览器开发者工具，在 Console 中粘贴即可连接并收发：

• const ws = new WebSocket("ws://localhost:8765");
• ws.onmessage = e => console.log("收到：", e.data);
• ws.onopen = () => ws.send("你好，服务端！");
• 看到控制台输出“收到：你好，服务端！”即表示通路正常

Python 客户端主动通信示例

适合做自动化测试或后台推送场景：

• 安装：pip install websocket-client
• 用 WebSocketApp 创建长连接，设置 on_message、on_open 回调
• 在 on_open 中调用 send() 发起首条消息
• 注意：阻塞式运行需调用 run_forever()，非阻塞可用 create_connection() + 循环 recv()

常见卡点与应对

初学者容易在这里停住：

• 跨域被拒：浏览器连接本地服务端时，WebSocket 协议本身不受同源策略限制，但服务端若校验 Origin 头则需手动放行
• 连接闪断：检查是否漏写 await websocket.recv() 导致协程退出；或客户端未正确处理 onclose
• 中文乱码：确保发送/接收都用字符串（不是 bytes），send("你好") 而非 send(b"\xe4\xbd\xa0\xe5\xa5\xbd")
• 无法广播：服务端需用 asyncio.wait(...) 并发向多个连接 send，不能直接 for 循环 await