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 PO设计模式的具体使用
这篇文章主要介绍了Python PO设计模式的具体使用，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2019-08-08
Python实战之Elasticsearch的高级实现详解
Elasticsearch是一个功能强大的开源搜索引擎,广泛应用于各种场景,本文将深入探讨如何使用Python与Elasticsearch进行高级实现,需要的可以参考下
2024-04-04
PyTorch两种安装方法
这篇文章主要介绍了PyTorch两种安装方法,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2021-03-03
PyQt5 designer 页面点击按钮跳转页面实现
本文主要介绍了PyQt5 designer 页面点击按钮跳转页面实现，文中通过示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2022-01-01
Python编程中被忽视的核心技巧总结
这篇文章主要介绍了一些在 Python 编程中可能被忽视的核心功能，包括默认参数、海象运算符、*args 和 **kwargs 的使用等，需要的可参考下
2023-08-08
python数据结构之二叉树的遍历实例
这篇文章主要介绍了python数据结构之二叉树的递归遍历实例,需要的朋友可以参考下
2014-04-04
用python 批量操作redis数据库
这篇文章主要介绍了如何用python 批量操作redis数据库，帮助大家更好的理解和学习使用python，感兴趣的朋友可以了解下
2021-03-03
Python深度学习pytorch神经网络多输入多输出通道
这篇文章主要为大家介绍了Python深度学习中pytorch神经网络多输入多输出通道的详解有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步
2021-10-10
Django实战之用户认证（初始配置）
这篇文章主要介绍了Django实战之用户认证（初始配置），小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2018-07-07
python使用logging模块记录日志的操作
在编写 Python 程序时,日志记录是一个非常重要的部分,日志不仅帮助你在开发过程中调试代码,还可以在程序正式运行时提供诊断信息,帮助定位问题,在本文中,我们将介绍 Python 的 logging 模块,帮助你了解如何使用这个模块来记录和管理日志,需要的朋友可以参考下
2024-10-10