网站导航
> 技术文档 > Python WebSockets 库详解：从基础到实战

Python WebSockets 库详解：从基础到实战

网友: 技术文档 2025-08-15 16:15:53

1. 引言

WebSocket 是一种全双工、持久化的网络通信协议，适用于需要低延迟的应用，如实时聊天、股票行情推送、在线协作、多人游戏等。相比传统的 HTTP 轮询方式，WebSocket 减少了带宽开销，提高了实时性。

在 Python 中，最流行的 WebSocket 库是 websockets，它是一个基于 asyncio 的轻量级 WebSocket 库，支持 WebSocket 服务器和客户端实现。本文将深入介绍 WebSockets 及其在 Python 中的使用方法。

2. 为什么使用 WebSocket？

在传统的 HTTP 轮询（Polling）或长轮询（Long Polling）中，客户端需要不断向服务器发送请求，即使没有数据更新，也会浪费带宽和资源。WebSocket 通过单次握手建立持久连接，服务器可以主动推送数据，极大地提高了通信效率。

WebSocket 的优势：

  • 低延迟：基于 TCP 连接，减少握手和数据传输时间。
  • 双向通信：服务器可以主动向客户端推送消息，而无需等待请求。
  • 减少带宽消耗：避免 HTTP 头部的额外开销，提高吞吐量。
  • 适用于实时应用：如聊天、直播、股票行情等。

3. 安装 WebSockets 库

首先，我们需要安装 websockets：

pip install websockets

websockets 依赖 Python 3.6 及以上版本，并且基于 asyncio，所以所有 WebSocket 代码都是**异步（async）**的。

4. 使用 WebSockets 搭建 WebSocket 服务器

WebSocket 服务器的基本实现只需几行代码。

4.1 WebSocket 服务器示例

import asyncioimport websocketsasync def echo(websocket, path): async for message in websocket: print(f\"收到消息: {message}\") await websocket.send(f\"服务器响应: {message}\")# 启动 WebSocket 服务器start_server = websockets.serve(echo, \"0.0.0.0\", 8765)asyncio.get_event_loop().run_until_complete(start_server)asyncio.get_event_loop().run_forever()

说明：

  • websockets.serve(echo, \"0.0.0.0\", 8765) 启动一个 WebSocket 服务器，监听 8765 端口。
  • async for message in websocket 监听客户端发送的消息，并在收到后回显给客户端。

5. WebSocket 客户端

WebSocket 客户端的实现也非常简单：

import asyncioimport websocketsasync def client(): async with websockets.connect(\"ws://localhost:8765\") as websocket: await websocket.send(\"Hello, WebSocket Server\") response = await websocket.recv() print(f\"服务器响应: {response}\")asyncio.run(client())

说明：

  • websockets.connect(\"ws://localhost:8765\") 连接 WebSocket 服务器。
  • await websocket.send(\"Hello, WebSocket Server\") 发送数据。
  • await websocket.recv() 接收服务器的消息。

6. 处理多个客户端

通常，我们需要处理多个客户端同时连接。在 WebSockets 中，可以使用 asyncio.gather() 来管理多个 WebSocket 连接。

6.1 广播消息给所有连接的客户端

import asyncioimport websocketsconnected_clients = set() # 记录已连接的客户端async def handler(websocket, path): connected_clients.add(websocket) try: async for message in websocket: print(f\"收到消息: {message}\") # 广播给所有客户端 await asyncio.gather(*(client.send(f\"广播消息: {message}\") for client in connected_clients)) finally: connected_clients.remove(websocket)start_server = websockets.serve(handler, \"0.0.0.0\", 8765)asyncio.get_event_loop().run_until_complete(start_server)asyncio.get_event_loop().run_forever()

说明：

  • 使用 connected_clients 集合存储所有连接的客户端。
  • 在 async for message in websocket 内部，遍历 connected_clients，将消息发送给所有客户端。

7. WebSocket 服务器的异常处理

实际应用中，客户端可能会断开连接，或者发送非法数据。我们需要在服务器端增加异常处理，以确保服务不会崩溃。

import asyncioimport websocketsasync def handler(websocket, path): try: async for message in websocket: print(f\"收到: {message}\") await websocket.send(f\"服务器回复: {message}\") except websockets.exceptions.ConnectionClosedError: print(\"客户端连接关闭\") except Exception as e: print(f\"发生错误: {e}\")start_server = websockets.serve(handler, \"0.0.0.0\", 8765)asyncio.get_event_loop().run_until_complete(start_server)asyncio.get_event_loop().run_forever()

8. 使用 WebSockets 传输 JSON 数据

在 WebSockets 通信中，通常需要传输结构化数据，例如 JSON。

服务器端：

import asyncioimport websocketsimport jsonasync def handler(websocket, path): async for message in websocket: data = json.loads(message) response = {\"message\": f\"收到: {data[\'content\']}\"} await websocket.send(json.dumps(response))start_server = websockets.serve(handler, \"0.0.0.0\", 8765)asyncio.get_event_loop().run_until_complete(start_server)asyncio.get_event_loop().run_forever()

客户端：

import asyncioimport websocketsimport jsonasync def client(): async with websockets.connect(\"ws://localhost:8765\") as websocket: data = json.dumps({\"content\": \"Hello, Server\"}) await websocket.send(data) response = await websocket.recv() print(f\"服务器响应: {json.loads(response)}\")asyncio.run(client())

9. WebSockets vs. HTTP

特性 WebSockets HTTP 连接方式 持久连接 请求-响应 数据推送 服务器主动推送 需要轮询 适用场景 实时应用（聊天、直播） 普通 Web API

10. WebSocket 实战：实时聊天室

import asyncioimport websocketsclients = set()async def chat(websocket, path): clients.add(websocket) try: async for message in websocket: await asyncio.gather(*(client.send(message) for client in clients)) finally: clients.remove(websocket)start_server = websockets.serve(chat, \"0.0.0.0\", 8765)asyncio.get_event_loop().run_until_complete(start_server)asyncio.get_event_loop().run_forever()

客户端可以连接服务器并发送消息，服务器会广播给所有连接的用户，形成一个实时聊天室。

总结

  • WebSocket 提供了低延迟、全双工通信，适用于实时应用。
  • websockets 库基于 asyncio，支持高并发通信。
  • WebSockets 可用于聊天系统、股票行情推送、多人协作、远程控制等应用场景。

通过本教程，你应该掌握了 Python websockets 库的使用方法，并能在项目中实现高效的实时通信！🚀

网络标签：

相关问题