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

· 日常编程 · 41596

WebSocket是一种全双工、持久化的网络通信协议,适用于需要低延迟的应用,如实时聊天、股票行情推送、在线协作、多人游戏等,本文给大家介绍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 实战：实时聊天室
            • 总结

              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库的使用指南
              • python中websockets与主线程传递参数的实现

              相关推荐：

              OHAUS电子天平EX124/AD串口通信故障排查与Python控制实战指南

              本文详解ohaus explorer ex124/ad电子天平通过rs232与树莓派5或windows pc通信时“能收不能发”的典型故障，揭示单向通信正常的usb转rs232线缆隐性缺陷，并提供经实测验证的python串口控制代码、关键配置要点及调试方法。 本文详解ohaus explorer ex124/ad电子天平通过rs232与树莓派5或windows pc通信时“能收不能发”的典型故障，...

              为什么Python 3.11的结构化模式匹配提升效率_实战演示match语法环境

              match语句通过字节码级优化和单次结构探测替代链式if判断，减少重复取值与类型检查，在Python 3.11+中实现更紧凑分支、更高JIT友好性及实际运行提速。 match 语句本身不直接“提升运行时效率”，但它能显著减少冗余判断、避免重复解包、降低嵌套 if 深度，从而让代码更接近数据结构本质——这带来的实际收益是：逻辑更清晰、分支更紧凑、JIT 友好（CPython 3.11 的自适应解释器...

              Python中相对导入与绝对导入怎么选_避免ImportError的实战指南

              必须用绝对导入当模块不在当前包结构内或脚本独立运行时，因相对导入依赖__package__属性且仅在包内导入有效；应统一用绝对导入并确保包根在sys.path中。 什么时候必须用绝对导入 当模块不在当前包结构内、或你正在写可独立运行的脚本（比如 python myscript.py）时，绝对导入是唯一可靠的选择。Python 的 __name__ 在直接运行时是 '__main__'，此时相对导入...

              CSS框架怎么布局_CSS框架选择与页面布局实战教程

              CSS框架通过预设的网格系统、响应式工具、UI组件和工具类，提升布局效率。其核心机制包括：基于Flexbox或CSS Grid的声明式网格系统，简化布局划分；内置断点与媒体查询实现响应式设计；封装常用组件保证UI一致性；提供原子化工具类增强样式灵活性。选择框架需权衡项目规模、团队熟悉度、定制需求与性能。常见问题如样式覆盖难、同质化、升级兼容性等，可通过变量定制、按需引入、PurgeCSS优化及结合...

              CSS怎么嵌套盒子_CSS盒子模型嵌套与布局实战教程

              嵌套盒子是HTML元素层级的体现，通过结构化HTML与CSS控制实现布局，如main-container包含header、sidebar和content，CSS定义各盒子样式与排列。 在CSS里，我们说的“嵌套盒子”其实就是HTML元素层级关系的直接体现。你把一个div放在另一个div里面，或者一个p标签嵌套在article里，这就是盒子嵌套。CSS的魔法在于，它能让你精妙地控制这些层层相套的盒子...

              “XX基础，XX就不基础”是什么梗？

              “XX基础，XX就不基础”源自电视剧《生万物》观众评论，是方言反语与幽默对比的结合，通过“基础”与“不基础”的反差强调事物间的依赖关系，如“生活方式基础，心血管健康就不基础”，因结构简单、易模仿、多场景适用且具情感共鸣而走红，广泛用于调侃或强调前提条件的重要性。 “XX基础，XX就不基础”是什么梗？ 最近，“XX基础，XX就不基础”这个句式在网络上流行起来，它用一种幽默反差的方式强调事物间的基础关...

              前端图片裁剪技术：Croppie.js实战指南

              本文详细介绍了如何在网页中实现交互式图片裁剪功能，主要利用JavaScript库Croppie.js结合HTML和CSS。教程涵盖了从环境搭建、HTML结构、CSS样式到JavaScript逻辑的完整实现过程，旨在帮助开发者在客户端高效地处理图片裁剪需求，提升用户体验。在现代web应用中，用户上传图片并进行裁剪已成为一项常见需求。无论是头像上传、商品图片编辑还是内容配图，提供一个直观、便捷的裁剪工...

              Sa-Token 基础及 Spring Boot 集成实战指南

              Sa-Token是轻量级Java权限认证框架,支持登录、权限、单点登录等场景,集成SpringBoot简便,提供丰富注解和模块化设计,相比Shiro和SpringSecurity更易用且适合中小型项目,注重安全与扩展性,本文给大家介绍Sa-Token基础及Spring Boot集成实战指南,感兴趣的一起看看吧 目录一、Sa-Token 简介二、核心概念1. 登录认证流程2. 核心组件3. 主要功能...

              Sentinel原理与SpringBoot整合实战案例讲解

              本文将详细介绍Sentinel的核心原理和基本功能点,并提供一个完整的SpringBoot整合Sentinel的案例,帮助读者快速掌握Sentinel的使用方法,感兴趣的朋友一起看看吧 目录前言一、Sentinel简介1.1 什么是Sentinel1.2 Sentinel与其他熔断降级工具的对比二、Sentinel核心原理2.1 基本工作原理2.2 Sentinel的执行流程2.3 Sentine...

              加速卷积函数的 Numba 优化实战

              本文旨在指导如何使用 Numba 优化卷积函数，通过将 NumPy 代码替换为显式循环，并利用 Numba 的并行化功能，显著提升代码执行效率。我们将深入探讨优化策略，并提供优化后的代码示例，最终实现比原始 NumPy 代码快数倍的加速效果。 优化思路：避免临时数组和利用显式循环 原始代码中使用了大量的 NumPy 操作，这在 Numba 中可能会导致性能瓶颈。Numba 更擅长处理显式循环，因为...