本文主要是介绍Python WebSockets 库从基础到实战使用举例，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
《PythonWebSockets库从基础到实战使用举例》WebSocket是一种全双工、持久化的网络通信协议,适用于需要低延迟的应用,如实时聊天、股票行情推送、在线协作、多人游戏等,本文给大家介...
1. 引言
WebSocket 是一种全双工、持久化的网络通信协议，适用于需要低延迟的应用，如实http://www.chinasem.cn时聊天、股票行情推送、在线协作、多人游戏等。相比传统的 HChina编程TTP 轮询方式，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 asyncio
import websockets
async 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 asyncio
import websockets
async 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 asyncio
import websockets
connected_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 asyncio
import websockets
async 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 数据
在 WebSockChina编程ets 通信中，通常需要传输结构化数据，例如 JSON。
服务器端：
import asyncio
import websockets
import json
async 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 asyncio
import websockets
import json
async 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 |
|---|---|---|
| 连接方式 | 持久连接 | 请求-响应 |
| 数据推送python | 服务器主动推送 | 需要轮询 |
| 适用场景 | 实时应用（聊天、直播） | 普通 Web API |
10. WebSocket 实战：实时聊天室
import asyncio
import websockets
clients = 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_serwww.chinasem.cnver)
asyncio.get_event_loop().run_forever()客户端可以连接服务器并发送消息，服务器会广播给所有连接的用户，形成一个实时聊天室。
总结
- WebSocket 提供了低延迟、全双工通信，适用于实时应用。
websockets库基于asyncio，支持高并发通信。- WebSockets 可用于聊天系统、股票行情推送、多人协作、远程控制等应用场景。
通过本教程，你应该掌握了 Python websockets 库的使用方法，并能在项目中实现高效的实时通信！
这篇关于Python WebSockets 库从基础到实战使用举例的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
