首页 > 脚本专栏 > python > Python WebSockets 库

Python WebSockets 库从基础到实战使用举例

2025-08-18 16:05:13 作者：萧鼎

WebSocket是一种全双工、持久化的网络通信协议,适用于需要低延迟的应用,如实时聊天、股票行情推送、在线协作、多人游戏等,本文给大家介绍Python WebSockets库从基础到实战使用举例,感兴趣的朋友一起看看吧

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 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 数据

在 WebSockets 通信中，通常需要传输结构化数据，例如 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
连接方式	持久连接	请求-响应
数据推送	服务器主动推送	需要轮询
适用场景	实时应用（聊天、直播）	普通 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_server)
asyncio.get_event_loop().run_forever()

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

总结

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

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

Python WebSockets 库从基础到实战使用举例

1. 引言

2. 为什么使用 WebSocket？

3. 安装 WebSockets 库

4. 使用 WebSockets 搭建 WebSocket 服务器

4.1 WebSocket 服务器示例

5. WebSocket 客户端

6. 处理多个客户端

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

7. WebSocket 服务器的异常处理

8. 使用 WebSockets 传输 JSON 数据

9. WebSockets vs. HTTP

10. WebSocket 实战：实时聊天室

总结

您可能感兴趣的文章: