Python 中如何实现 WebSocket 通信

WebSocket 通信在现代 Web 开发中扮演着至关重要的角色，它提供了一种实时、双向的通信方式，适用于许多应用场景，从在线聊天到实时数据更新。Python 作为一种多用途的编程语言，也支持轻松实现 WebSocket 通信。本文将深入介绍 Python 中如何实现 WebSocket 通信，为你介绍使用场景、多种方法和实践案例。

使用场景

WebSocket 通信适用于需要实时更新或与服务器进行双向通信的场景。一些常见的使用场景包括：

在线聊天应用：实时性是聊天应用的关键，WebSocket 可以提供快速的消息传递。
实时数据监控：用于实时监控系统状态、数据变化等。
在线游戏：支持实时多人游戏通信，提升游戏体验。
协作工具：用于实时协作编辑、共享文档等场景。

WebSocket 调试工具

要调试 WebSocket，那就需要一个好的调试工具，这里我比较推荐 Apifox。它支持调试 http（s）、WebSocket、Socket、gRPC、Dubbo 等多种协议的接口，这使得它成为了一个非常全面的接口测试工具！

立即体验 Apifox

常见方法

以下是一些在 Python 中实现 WebSocket 通信的方法。

方法 1：使用 websockets 库

websockets 是一个简单而强大的 Python 库，用于 WebSocket 通信。以下是一个简单的例子：

import asyncio
import websockets

async def echo(websocket, path):
    async for message in websocket:
        await websocket.send(message)

start_server = websockets.serve(echo, "localhost", 8765)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

方法 2：使用 socket.io 库

socket.io 是一个流行的跨平台库，支持 WebSocket 通信。安装：

pip install python-socketio

使用示例：

import socketio

sio = socketio.Server()

app = socketio.WSGIApp(sio)

@sio.event
def connect(sid, environ):
    print(f"Connection {sid} connected")

@sio.event
def disconnect(sid):
    print(f"Connection {sid} disconnected")

if __name__ == '__main__':
    import eventlet
    eventlet.wsgi.server(eventlet.listen(('localhost', 5000)), app)

方法 3：使用 FastAPI 框架

FastAPI 是一个现代、快速（基于 Starlette 和 Pydantic）的 Web 框架，也支持 WebSocket。安装：

pip install fastapi uvicorn

使用示例：

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

其他常见方法

除了上述方法，还有一些其他库和框架，如aiohttp、tornado等，可以根据项目需求选择合适的工具。

实践案例

为了更好地理解在 Python 中如何实现 WebSocket 通信，让我们通过一个详细的实践案例来演示整个过程。在这个案例中，我们将创建一个简单的聊天应用，其中包括 WebSocket 服务器和客户端的实现。

步骤 1：安装必要的库

首先，我们需要安装websockets库，它是一个用于 WebSocket 通信的简单而强大的库。使用以下命令安装：

pip install websockets

步骤 2：WebSocket 服务器的实现

创建一个名为websocket_server.py的文件，并添加以下代码：

import asyncio
import websockets

async def chat_server(websocket, path):
    async for message in websocket:
        # 接收客户端发送的消息
        print(f"Received message: {message}")

        # 将消息发送给所有连接的客户端
        await asyncio.gather(*[client.send(message) for client in clients])

# 存储所有连接的客户端
clients = set()

start_server = websockets.serve(chat_server, "localhost", 3000)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

在这个服务器代码中，我们创建了一个chat_server协程，用于处理与客户端的通信。它会接收客户端发送的消息，并将消息广播给所有连接的客户端。

步骤 3：运行服务器

现在，我们可以同时运行服务器和多个客户端以测试我们的 WebSocket 通信。在两个不同的终端窗口中运行以下命令：

python websocket_server.py

通过在客户端输入消息，你将看到服务器接收并广播消息给所有连接的客户端。

步骤 5：使用 Apifox 调试

WebSocket 如果你要调试 WebSocket 接口，并确保你的应用程序能够正常工作。这时，一个强大的接口测试工具就会派上用场。

Apifox 是一个比 Postman 更强大的接口测试工具，Apifox = Postman + Swagger + Mock + JMeter。它支持调试 http（s）、WebSocket、Socket、gRPC、Dubbo 等多种协议的接口，这使得它成为了一个非常全面的接口测试工具，所以强烈推荐去下载体验！

首先在 Apifox 中新建一个 HTTP 项目，然后在项目中添加 WebSocket 接口。

立即体验 Apifox