主流 Python WebSocket 库概览

websockets - 异步高性能首选

websockets 是一个为 Python 异步编程（asyncio）设计的库，是构建 WebSocket 服务器和客户端的事实标准，它性能优异，功能全面,是大多数生产环境下的首选。

安装

pip install websockets

核心特点

• 完全异步：基于 asyncio,能高效处理大量并发连接。
• 双向通信：服务器和客户端都可以主动发送和接收消息。
• 支持 RFC 6455：完全兼容 WebSocket 标准。
• 客户端和服务器端：同一个库可以同时用于编写客户端和服务器端代码。

代码示例

A. WebSocket 服务器

# server.py
import asyncio
import websockets
# 当有新的客户端连接时，此函数会被调用
async def handler(websocket, path):
    """
    path 是客户端连接的路径，通常用于路由
    """
    print(f"新的客户端已连接: {websocket.remote_address}")
    try:
        # 持续接收来自客户端的消息
        async for message in websocket:
            print(f"收到消息: {message}")
            # 将收到的消息原样返回给客户端
            await websocket.send(f"服务器已收到你的消息: {message}")
    except websockets.exceptions.ConnectionClosed as e:
        print(f"客户端 {websocket.remote_address} 已断开连接: {e}")
# 启动 WebSocket 服务器
start_server = websockets.serve(handler, "localhost", 8765)
# 运行服务器，直到被手动停止
asyncio.get_event_loop().run_until_complete(start_server)
print("WebSocket 服务器已启动，监听 ws://localhost:8765")
asyncio.get_event_loop().run_forever()

B. WebSocket 客户端

# client.py
import asyncio
import websockets
async def hello():
    uri = "ws://localhost:8765"
    async with websockets.connect(uri) as websocket:
        print("已连接到服务器")
        # 发送一条消息
        name = input("请输入你的名字: ")
        await websocket.send(f"你好，服务器！我是 {name}.")
        # 等待并打印服务器的回复
        response = await websocket.recv()
        print(f"服务器回复: {response}")
# 运行客户端
asyncio.get_event_loop().run_until_complete(hello())

如何运行:

1. 先运行 server.py。
2. 再运行 client.py（可以运行多个客户端实例）。
3. 在客户端的终端输入内容,服务器会收到并回复。

python-socketio - 快速开发与兼容性首选

如果你需要与使用 Socket.IO 的客户端（如 JavaScript 客户端）通信，或者希望快速搭建一个功能丰富的实时应用（如房间、事件命名、自动重连等），python-socketio 是最佳选择。

安装

pip install "python-socketio[client]" # 如果只需要客户端
pip install "python-socketio[asyncio_server]" # 如果只需要异步服务器
pip install "python-socketio[asyncio_client]" # 如果只需要异步客户端
pip install "python-socketio[asyncio]" # 同时安装服务器和客户端

核心特点

• Socket.IO 协议：支持 Socket.IO 协议，可以与浏览器、移动端的 Socket.IO 客户端无缝对接。
• 多种传输：在 WebSocket 不可用时,会自动降级到轮询等其他方式。
• 高级功能：内置了“房间”（Rooms）、“命名事件”（Named Events）、“广播”（Broadcasting）等高级功能,开箱即用。
• 同步和异步支持：既支持同步（基于 eventlet 或 gevent），也支持异步（基于 asyncio）。

代码示例 (基于 asyncio)

A. Socket.IO 服务器

# sio_server.py
import asyncio
import socketio
# 创建一个异步的 Socket.IO 客户端管理器
sio = socketio.AsyncServer(async_mode='asyncio', cors_allowed_origins='*')
# 当客户端连接时触发
@sio.event
async def connect(sid, environ):
    print(f'客户端 {sid} 已连接')
    await sio.emit('message', {'data': '欢迎连接！'}, room=sid)
# 当客户端断开连接时触发
@sio.event
async def disconnect(sid):
    print(f'客户端 {sid} 已断开')
# 当收到 'my_message' 事件时触发
@sio.event
async def my_message(sid, data):
    print(f'从客户端 {sid} 收到消息: {data}')
    # 将消息广播给所有客户端（除了发送者）
    await sio.emit('message', {'data': data}, skip_sid=sid)
# 将 ASGI 应用包装起来，以便可以被 ASGI 服务器（如 Uvicorn）运行
app = socketio.ASGIApp(sio)
if __name__ == '__main__':
    # 使用 Uvicorn 运行 ASGI 应用
    import uvicorn
    uvicorn.run(app, host="localhost", port=8000)

B. Socket.IO 客户端

# sio_client.py
import socketio
# 创建一个异步的 Socket.IO 客户端
sio = socketio.AsyncClient()
# 当连接到服务器时触发
@sio.event
async def connect():
    print('已成功连接到服务器！')
    # 发送一个自定义事件
    await sio.emit('my_message', {'data': '这是来自客户端的消息！'})
# 当收到 'message' 事件时触发
@sio.event
async def message(data):
    print('收到服务器消息:', data)
# 当断开连接时触发
@sio.event
async def disconnect():
    print('已断开连接')
# 启动客户端并连接服务器
async def main():
    await sio.connect('http://localhost:8000')
    await sio.wait() # 保持连接
if __name__ == '__main__':
    asyncio.run(main())

SimpleWebSocketServer - 简单入门首选

如果你只是想快速理解 WebSocket 的工作原理，或者需要一个简单的同步服务器来推送数据（IoT 设备状态），SimpleWebSocketServer 是一个非常不错的选择，它不依赖 asyncio,代码非常直观。

安装

pip install SimpleWebSocketServer

核心特点

• 同步模型：代码逻辑简单，易于理解,适合初学者。
• 轻量级：没有复杂的依赖。
• 易于扩展：通过继承 WebSocket 类可以轻松自定义处理逻辑。

代码示例

# simple_server.py
from SimpleWebSocketServer import SimpleWebSocketServer, WebSocket
# 定义一个自定义的 WebSocket 处理类
class SimpleEcho(WebSocket):
    # 当客户端连接时调用
    def handleConnected(self):
        print(f'新客户端连接: {self.address[0]}:{self.address[1]}')
        self.sendMessage('欢迎连接到简单回显服务器！')
    # 当客户端发送消息时调用
    def handleMessage(self):
        # self.data 包含客户端发送的消息
        print(f'收到消息: {self.data}')
        self.sendMessage(self.data) # 将消息回显给客户端
    # 当客户端断开连接时调用
    def handleClose(self):
        print(f'客户端 {self.address[0]}:{self.address[1]} 已断开连接')
# 启动服务器
server = SimpleWebSocketServer('', 8000, SimpleEcho)
print('简单 WebSocket 服务器已启动，监听 ws://localhost:8000')
server.serveforever()

Django Channels - Django 生态集成

如果你的项目是基于 Django 的，并且需要添加实时功能（如在线聊天、实时通知、协作编辑等），Django Channels 是不二之选，它将 Django 的请求/响应模型扩展到了异步的 WebSocket 连接。

安装

pip install channels channels_redis

核心特点

• Django 无缝集成：将 WebSocket 请求当作一种新的“路由”来处理，与 Django 的视图系统非常相似。
• 层叠架构：Channel Layer 允许不同的消费者（Consumer）实例之间进行通信,非常适合分布式部署。
• 协议支持：除了 WebSocket，还支持 HTTP/2、MQTT 等。

工作流程简介

1. ASGI 服务器：接收 WebSocket 请求（如 Daphne）。
2. 路由层：将请求根据路径分发给对应的 消费者。
3. 消费者：处理 WebSocket 事件（connect, disconnect, receive），并通过 通道层 与其他消费者或后台任务通信。
4. 通道层：通常由 Redis 或 RabbitMQ 实现,负责进程间通信。

如何选择？一张图帮你决策

graph TD
    A[我需要用 Python 开发 WebSocket 应用] --> B{你的主要需求是什么?};
    B --> C[构建高性能、异步的后端服务，如聊天室、游戏];
    B --> D[需要与 JavaScript 的 Socket.IO 客户端通信];
    B --> E[只是想快速学习 WebSocket 原理或做个简单的数据推送];
    B --> F[我的项目是 Django，需要添加实时功能];
    C --> G[选择 <b>websockets</b><br/> (异步、高性能、标准)];
    D --> H[选择 <b>python-socketio</b><br/> (协议兼容、功能丰富)];
    E --> I[选择 <b>SimpleWebSocketServer</b><br/> (同步、简单、易上手)];
    F --> J[选择 <b>Django Channels</b><br/> (与 Django 深度集成)];

• websockets：现代 Python 异步编程的基石，功能强大，性能卓越,是大多数新项目的首选。
• python-socketio：如果你在构建一个生态系统，需要与多种客户端（尤其是前端）交互，或者需要快速利用房间、广播等高级功能,选它没错。
• SimpleWebSocketServer：学习、教学或构建极简原型时的利器。
• Django Channels：Django 开发者的不二之选，为你的 Django 应用注入实时能力。

对于绝大多数新的 Python WebSocket 项目，我都会从 websockets 开始，除非有明确的 Socket.IO 兼容性需求。