使用 websockets 库可快速启动最小 WebSocket 服务:安装 pip install websockets,定义异步 echo 函数处理 recv/send,调用 serve(echo, "localhost", 8765) 并用 asyncio.run() 启动;需注意连接隔离、异常捕获、心跳响应及事件循环兼容性。

用 websockets 库启动一个最小可运行服务
Python 3.10 本身不带 WebSocket 服务器支持,必须依赖第三方库;websockets 是目前最轻量、最贴近协议规范的选择,比 Flask-SocketIO 或 FastAPI + websockets 更少抽象层,适合快速验证逻辑。安装后只需定义一个异步处理函数,再调用 serve() 即可监听连接。
实操建议:
- 执行
pip install websockets(注意不是websocket,后者是客户端库) - 服务端代码不超过 10 行:定义
echo异步函数接收websocket对象,用await websocket.recv()读、await websocket.send()写 - 启动时传入
start_server = serve(echo, "localhost", 8765),再asyncio.get_event_loop().run_until_complete(start_server) - 别漏掉
asyncio.run()或事件循环管理——Python 3.10 默认不自动运行协程,直接调用会报RuntimeWarning: coroutine 'serve' was never awaited
处理多个客户端时避免消息错乱
默认的 echo 示例只对单个连接生效,但真实场景中多个浏览器标签页或设备会同时连上来,websocket 对象彼此独立,不能共用变量存储状态。常见错误是把用户数据存在全局 dict 里却没加连接生命周期钩子,导致断开后残留、新连接覆盖旧数据。
实操建议:
- 每个连接应有唯一标识:可用
id(websocket)或hash(websocket)作临时 key,更稳妥的是在握手阶段解析websocket.request_headers.get("Sec-WebSocket-Key") - 用
async with或try/finally确保连接关闭时清理对应状态,例如del clients[client_id] - 广播消息不要遍历所有
websocket对象后直接await send()—— 这会阻塞其他连接;改用asyncio.create_task(websocket.send(msg))并发发出 - 若需跨连接共享状态(如房间名、在线人数),用
asyncio.Lock()保护共享变量,否则并发写入可能丢数据
调试连接失败或 ConnectionClosedError
前端 new WebSocket("ws://localhost:8765") 打不开,或刚发一条消息就断开,多数不是协议问题,而是 Python 侧未正确响应 ping/pong 或异常未捕获。浏览器会静默关闭无心跳的连接,而 websockets 默认每 20 秒发一次 ping,但若 handler 中抛出未处理异常(比如 JSONDecodeError),连接立即终止并报 ConnectionClosedError: code = 1006 (connection closed abnormally)。
实操建议:
- 在主 handler 外层加
try/except Exception as e:,至少print(f"Error from {websocket.remote_address}: {e}"),否则异常被吞掉,日志里看不到线索 - 确认 URL 协议是
ws://(开发用),不是http://或wss://(后者需要 TLS 证书) - 检查防火墙或 IDE 自带的端口占用提示——Python 3.10 在 macOS 上有时会因权限问题无法绑定低端口,换 8765 以上更稳妥
- 用
curl -i http://localhost:8765测试 HTTP 路由会返回 426 Upgrade Required,这是正常行为;若返回 404 或连接拒绝,说明服务根本没起来
与 asyncio 事件循环的兼容性细节
Python 3.10 对 asyncio 做了若干调整,比如默认事件循环改为 ProactorEventLoop(Windows)或 SelectorEventLoop(Linux/macOS),而 websockets 依赖底层 socket 的异步读写能力。某些旧代码用 loop.run_forever() 启动,但在 3.10+ 可能触发 RuntimeError: This event loop is already running,尤其和 Jupyter 或某些测试框架混用时。
实操建议:
- 坚持用
asyncio.run(main())启动顶层协程,不要手动调用get_event_loop()—— Python 3.10 已明确不鼓励该用法 - 若需嵌入现有事件循环(如 FastAPI 的 lifespan hook),用
asyncio.create_task()而非loop.create_task() -
websockets.serve()返回的是Server实例,不是协程,不能直接await;正确写法是server = await serve(...),然后await server.wait_closed()等待停止 - 别在 handler 里调用阻塞函数(如
time.sleep()、requests.get())——必须用await asyncio.sleep()或await loop.run_in_executor()包装,否则整个服务卡死
真正麻烦的不是启动服务,而是连接生命周期管理——从握手、心跳、消息解析到异常恢复,每一步都得显式编码。很多人卡在“为什么前端连上了却收不到回包”,其实只是忘了 await 那个 send() 调用。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











