WebSocket 是一种全双工通信协议,允许在客户端和服务器之间进行实时数据交换。相比于传统的 HTTP 协议,它在需要频繁数据更新的场景(如聊天应用、实时通知、在线游戏等)中具有更高的效率。
Python 提供了多个库来实现 WebSocket 通信,包括 websockets
、Socket.IO
、Tornado
等。以下是使用 websockets
库进行 WebSocket 通信的示例。
websockets
pip install websockets
服务器端代码将侦听来自客户端的 WebSocket 连接,并处理消息。
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}")
async def main():
server = await websockets.serve(echo, "localhost", 8765)
await server.wait_closed()
asyncio.run(main())
websockets.serve(echo, "localhost", 8765)
:启动 WebSocket 服务器,监听 localhost
上的 8765 端口。async for message in websocket
:处理从客户端接收到的消息,并通过 await websocket.send(...)
将其回发给客户端。客户端代码将连接到 WebSocket 服务器,并发送和接收消息。
import asyncio
import websockets
async def hello():
uri = "ws://localhost:8765"
async with websockets.connect(uri) as websocket:
await websocket.send("Hello, WebSocket!")
response = await websocket.recv()
print(f"Received response: {response}")
asyncio.run(hello())
websockets.connect(uri)
:连接到指定的 WebSocket 服务器。await websocket.send("Hello, WebSocket!")
:发送消息到服务器。await websocket.recv()
:接收服务器的响应消息。Socket.IO
实现 WebSocketSocket.IO
是一个更高层次的库,支持回退机制,可以在 WebSocket 不可用时自动回退到其他协议如长轮询。以下是使用 python-socketio
实现 WebSocket 的示例。
python-socketio
pip install python-socketio
import socketio
sio = socketio.AsyncServer(async_mode='asgi')
app = socketio.ASGIApp(sio)
@sio.event
async def connect(sid, environ):
print('Client connected:', sid)
await sio.emit('message', 'Welcome!', to=sid)
@sio.event
async def message(sid, data):
print('Message from client:', data)
await sio.emit('message', f"Echo: {data}", to=sid)
@sio.event
async def disconnect(sid):
print('Client disconnected:', sid)
if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host='localhost', port=8765)
socketio.AsyncServer(async_mode='asgi')
:创建一个异步 Socket.IO 服务器,使用 ASGI 兼容的应用。sio.event
:装饰器用于注册事件处理器,如 connect
、message
、disconnect
。import socketio
sio = socketio.AsyncClient()
@sio.event
async def connect():
print("Connected to server")
await sio.send("Hello, Socket.IO!")
@sio.event
async def message(data):
print("Message from server:", data)
@sio.event
async def disconnect():
print("Disconnected from server")
async def main():
await sio.connect('http://localhost:8765')
await sio.wait()
import asyncio
asyncio.run(main())
sio.send("Hello, Socket.IO!")
:发送消息到服务器。sio.connect(...)
:连接到服务器,并处理各种事件。Django Channels
实现 WebSocket对于使用 Django 构建的 Web 应用,你可以使用 Django Channels
来处理 WebSocket 连接。
Django Channels
pip install channels
修改 settings.py
以使用 Channels:
INSTALLED_APPS = [
# 其他应用
'channels',
]
ASGI_APPLICATION = 'myproject.asgi.application'
创建 asgi.py
文件:
import os
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application
from channels.auth import AuthMiddlewareStack
from channels.security.websocket import AllowedHostsOriginValidator
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
application = ProtocolTypeRouter({
"http": get_asgi_application(),
"websocket": AllowedHostsOriginValidator(
AuthMiddlewareStack(
URLRouter(
# 在这里添加路由
)
)
),
})
创建一个 WebSocket 消费者处理连接和消息:
from channels.generic.websocket import AsyncWebsocketConsumer
import json
class ChatConsumer(AsyncWebsocketConsumer):
async def connect(self):
await self.accept()
async def disconnect(self, close_code):
pass
async def receive(self, text_data):
text_data_json = json.loads(text_data)
message = text_data_json['message']
await self.send(text_data=json.dumps({
'message': message
}))
在 routing.py
中配置路由:
from django.urls import path
from . import consumers
websocket_urlpatterns = [
path('ws/chat/', consumers.ChatConsumer.as_asgi()),
]
websockets
库 提供了基础的 WebSocket 支持,适合简单应用。Socket.IO
具有回退机制,适合需要高兼容性的应用。Django Channels
为 Django 项目提供了强大的 WebSocket 支持,适合构建复杂的实时 Web 应用。通过这些工具和库,你可以轻松地在 Python 应用中实现 WebSocket 实时通信功能。