在 Python 中，`websockets` 是一个基于 `asyncio` 的库，用于构建 WebSocket 服务器和客户端。它支持异步编程，非常适合用于需要实时通信的应用场景，例如在线聊天、实时数据推送、物联网（IoT）等[^2]。以下是使用 `websockets` 的基本教程。 ### 安装 在使用 `websockets` 库之前，需要先安装。可以通过 `pip` 进行安装： ```bash pip install websockets ``` ### 创建 WebSocket 服务器 以下是一个简单的 WebSocket 服务器示例，该服务器接收客户端发送的消息，并将其回传给客户端： ```python import asyncio import websockets async def echo(websocket, path): async for message in websocket: print(f"Received message: {message}") await websocket.send(f"Echo: {message}") start_server = websockets.serve(echo, "localhost", 8765) asyncio.get_event_loop().run_until_complete(start_server) asyncio.get_event_loop().run_forever() ``` 在这个示例中，`echo` 函数是一个处理 WebSocket 连接的协程。每当有客户端连接时，它会异步地接收消息，并通过 `websocket.send()` 方法将收到的消息回传给客户端。 ### 创建 WebSocket 客户端 以下是一个简单的 WebSocket 客户端示例，该客户端连接到服务器并发送一条消息： ```python import asyncio import websockets async def hello(): uri = "ws://localhost:8765" async with websockets.connect(uri) as websocket: await websocket.send("Hello, server!") response = await websocket.recv() print(f"Received from server: {response}") asyncio.get_event_loop().run_until_complete(hello()) ``` 在这个示例中，客户端通过 `websockets.connect()` 方法连接到指定的 URI，然后使用 `websocket.send()` 方法向服务器发送消息，并通过 `websocket.recv()` 方法接收服务器的响应。 ### 处理错误 在实际应用中，可能会遇到各种错误，例如连接失败或消息接收失败。可以通过 `try-except` 块来捕获并处理这些异常： ```python async def safe_hello(): uri = "ws://localhost:8765" try: async with websockets.connect(uri) as websocket: await websocket.send("Hello, server!") response = await websocket.recv() print(f"Received from server: {response}") except websockets.exceptions.ConnectionClosed as e: print(f"Connection closed: {e}") except Exception as e: print(f"An error occurred: {e}") asyncio.get_event_loop().run_until_complete(safe_hello()) ``` 在这个示例中，`websockets.exceptions.ConnectionClosed` 异常用于捕获连接关闭的情况，而其他异常则通过通用的 `Exception` 类捕获。 ### 实际应用场景 `websockets` 库可以用于多种实时通信场景，例如： - **在线聊天应用**：实现客户端与服务器之间的实时消息传递。 - **实时数据推送**：如股票信息、天气信息、新闻更新等。 - **物联网（IoT）**：用于服务器与设备间的实时通信。 - **在线教育或在线视频类应用**：老师可以实时看到所有在线学生的反馈。 - **实时监控**：例如流量监控、系统性能监控等需要实时反馈数据的场景[^3]。 ### 最佳实践 - **使用异步编程**：充分利用 `asyncio` 的优势，编写高效的异步代码。 - **处理连接和消息的异常**：确保在连接失败或消息接收失败时能够优雅地处理错误。 - **优化性能**：根据应用的需求，合理配置 WebSocket 的参数，例如超时时间、缓冲区大小等。 - **安全性**：在生产环境中，建议使用 `wss://`（WebSocket Secure）协议来加密通信，确保数据的安全性。 通过这些基本的示例和最佳实践，可以快速上手 `websockets` 库，并构建高效的实时通信应用。 ---