PythonWebSocket教程_实时通信基础实践

10次阅读

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

PythonWebSocket教程_实时通信基础实践

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 客户端主动通信示例

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

立即学习Python免费学习笔记(深入)”;

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

常见卡点与应对

初学者容易在这里停住:

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