当前位置： 技术文章>> Python 如何操作 WebSocket 连接？
文章标题：Python 如何操作 WebSocket 连接？
在Python中操作WebSocket连接是一项常见且强大的功能，它允许开发者实现全双工通信，即数据可以在客户端和服务器之间双向实时流动。WebSocket协议建立在HTTP协议之上，但一旦连接建立，就会变为一个独立的TCP连接，并绕过HTTP的头部和状态码，从而极大地减少数据传输的开销。接下来，我们将深入探讨如何在Python中使用WebSocket，并介绍几个流行的库，如`websockets`和`socket.io`，以及如何在你的项目中集成它们。
### WebSocket基础
WebSocket协议通过单个TCP连接提供了从服务器推送数据到客户端的能力，这对于实时应用如在线游戏、聊天应用、实时通知系统等至关重要。WebSocket连接通常通过HTTP协议的"Upgrade"请求来初始化，之后客户端和服务器之间的数据交换就不再依赖于HTTP。
### Python中的WebSocket库
在Python中，有多个库可以帮助我们实现WebSocket的客户端和服务器功能。这里主要介绍两个流行的库：`websockets`和`python-socketio`。
#### 1. 使用`websockets`库
`websockets`是一个轻量级的、符合WebSocket协议规范的Python库，它易于使用且功能强大。下面分别介绍如何使用`websockets`创建WebSocket服务器和客户端。
##### WebSocket服务器
首先，你需要安装`websockets`库（如果尚未安装）：
```bash
pip install websockets
```
然后，可以编写一个简单的WebSocket服务器：
```python
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()
```
这段代码创建了一个简单的WebSocket服务器，它会回显从客户端接收到的所有消息。
##### 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 world!")
response = await websocket.recv()
print(f"Received: {response}")
asyncio.get_event_loop().run_until_complete(hello())
```
客户端连接到服务器，发送一条消息，并打印服务器的回显响应。
#### 2. 使用`python-socketio`库
`python-socketio`是一个更高级的库，它建立在`Flask`或`Django`等Web框架之上，提供了对WebSocket和HTTP长轮询的双重支持。这意味着它不仅可以用于WebSocket通信，还能在WebSocket不可用时优雅地降级到长轮询。
##### 安装
首先，安装`python-socketio`及其依赖的`Flask`或`Django`（取决于你的项目需求）：
```bash
pip install python-socketio[asyncio_server]
pip install Flask # 如果你选择使用Flask
```
##### WebSocket服务器（基于Flask）
```python
from flask import Flask, render_template
from flask_socketio import SocketIO, send
app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app, async_mode='asyncio')
@app.route('/')
async def index():
return await render_template('index.html')
@socketio.on('message')
async def handleMessage(msg):
print('Message: ' + msg)
await send(f'Message from server: {msg}')
if __name__ == '__main__':
socketio.run(app)
```
这段代码创建了一个基于Flask的WebSocket服务器，它监听`message`事件，并将接收到的消息回发给客户端。
##### WebSocket客户端（HTML + JavaScript）
由于`python-socketio`库主要服务于服务器端，客户端通常使用JavaScript通过浏览器进行连接。但为了完整性，这里提供一个简单的HTML和JavaScript示例：
```html
SocketIO Test