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

样发回






