Python WebSocket[编辑 | 编辑源代码]

WebSocket 是一种在单个 TCP 连接上进行全双工通信的协议，允许服务器和客户端之间实时交换数据。Python 提供了多种库（如 `websockets` 和 `socket.io`）来实现 WebSocket 功能，适用于实时聊天、在线游戏、股票行情推送等场景。

概述[编辑 | 编辑源代码]

WebSocket 协议（RFC 6455）在 HTTP 握手后升级为持久连接，克服了传统 HTTP 轮询的低效问题。其核心特点包括：

双向通信：客户端和服务器可主动发送消息。
低延迟：无需重复建立连接。
轻量级：数据帧头部开销小（仅 2-10 字节）。

与 HTTP 对比[编辑 | 编辑源代码]

特性	WebSocket	HTTP
连接类型	持久化	短连接
数据流向	双向	单向（请求-响应）
适用场景	实时应用	静态资源获取

Python 实现[编辑 | 编辑源代码]

Python 主流库包括：

```
websockets
```
（异步）
```
Flask-SocketIO
```
（基于 Flask）
```
Django Channels
```
（集成 Django）

基础示例[编辑 | 编辑源代码]

使用

websockets

库实现简易服务端和客户端：

服务端[编辑 | 编辑源代码]

import asyncio
import websockets

async def echo(websocket, path):
    async for message in websocket:
        print(f"Received: {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()

客户端[编辑 | 编辑源代码]

import asyncio
import websockets

async def send_message():
    async with websockets.connect("ws://localhost:8765") as ws:
        await ws.send("Hello WebSocket!")
        response = await ws.recv()
        print(f"Server response: {response}")

asyncio.get_event_loop().run_until_complete(send_message())

输出：

# 服务端
Received: Hello WebSocket!

# 客户端
Server response: Echo: Hello WebSocket!

高级特性[编辑 | 编辑源代码]

消息序列化[编辑 | 编辑源代码]

常用 JSON 格式传输结构化数据：

import json

# 发送端
await websocket.send(json.dumps({"user": "Alice", "action": "join"}))

# 接收端
data = json.loads(await websocket.recv())

错误处理[编辑 | 编辑源代码]

捕获连接异常：

try:
    async with websockets.connect("ws://invalid.url") as ws:
        await ws.send("test")
except websockets.exceptions.ConnectionClosedError:
    print("Connection closed abnormally!")

实际案例[编辑 | 编辑源代码]

实时聊天室[编辑 | 编辑源代码]

实现代码片段：

connected_clients = set()

async def chat_handler(websocket, path):
    connected_clients.add(websocket)
    try:
        async for message in websocket:
            for client in connected_clients:
                if client != websocket:
                    await client.send(f"User: {message}")
    finally:
        connected_clients.remove(websocket)