深入理解FastAPIWebSocket

深入理解 FastAPI WebSocket

FastAPI 凭借其高性能、易用性和基于标准类型提示的特性，迅速成为构建 API 的热门框架。而在现代 Web 应用中，实时通信的需求日益增长，WebSocket 作为实现双向通信的协议，也越来越受到开发者的青睐。本文将深入探讨 FastAPI 中 WebSocket 的使用，帮助你构建强大的实时应用。

一、WebSocket 简介

WebSocket 是一种在单个 TCP 连接上进行全双工通信的协议。与传统的 HTTP 请求-响应模式不同，WebSocket 允许服务器主动向客户端推送数据，从而实现实时更新和交互。

主要特点：

• 双向通信： 服务器和客户端可以同时发送和接收数据。
• 持久连接： 建立连接后，连接保持打开状态，直到一方主动关闭。
• 低延迟： 减少了 HTTP 的握手开销，数据传输更高效。
• 减少带宽： 避免了不必要的 HTTP 请求，节省带宽资源。

适用场景：

• 在线聊天室
• 多人游戏
• 实时数据更新（例如：股票行情、体育赛事比分）
• 协同编辑
• 物联网设备监控

二、FastAPI 中 WebSocket 的基本使用

FastAPI 基于 Starlette 框架，对 WebSocket 提供了简洁而强大的支持。

1. 安装必要的库

bash
pip install fastapi uvicorn websockets

2. 简单的 WebSocket 示例

```python
from fastapi import FastAPI, WebSocket
import uvicorn

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")

if name == "main":
uvicorn.run(app, host="0.0.0.0", port=8000)
```

代码解释：

• @app.websocket("/ws")：使用装饰器将函数定义为 WebSocket 端点，指定路由 /ws。
• websocket: WebSocket：函数参数声明为 WebSocket 类型，用于操作 WebSocket 连接。
• await websocket.accept()：接受客户端的 WebSocket 连接请求。
• await websocket.receive_text()：接收客户端发送的文本数据。
• await websocket.send_text()：向客户端发送文本数据。

3. 运行和测试

运行上述代码后，可以使用 JavaScript 或其他 WebSocket 客户端工具连接到 ws://localhost:8000/ws 进行测试。

三、深入理解 WebSocket 交互

1. 连接的建立与关闭

• 建立连接： 客户端发起 WebSocket 连接请求，服务器端通过 await websocket.accept() 接受连接，建立双向通信通道。
• 关闭连接： 可以通过 await websocket.close() 主动关闭连接，或者客户端断开连接。关闭连接时，服务器端会收到 WebSocketDisconnect 异常。

2. 数据收发

FastAPI 支持接收和发送不同类型的数据：

• 文本数据： await websocket.receive_text() 和 await websocket.send_text()
• 二进制数据： await websocket.receive_bytes() 和 await websocket.send_bytes()
• JSON 数据： await websocket.receive_json() 和 await websocket.send_json()

3. 异常处理

在 WebSocket 连接过程中，可能会遇到各种异常情况，例如客户端断开连接、网络错误等。需要使用 try...except 块来捕获和处理这些异常，确保程序的健壮性。

```python
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import uvicorn

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")
except WebSocketDisconnect:
print("Client disconnected")
except Exception as e:
print(f"An error occurred: {e}")

if name == "main":
uvicorn.run(app, host="0.0.0.0", port=8000)
```

四、高级用法

1. 广播消息

可以维护一个连接池，向所有已连接的客户端广播消息。

```python
from fastapi import FastAPI, WebSocket
import uvicorn

app = FastAPI()

connected_clients = set()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
connected_clients.add(websocket)
try:
while True:
data = await websocket.receive_text()
for client in connected_clients:
await client.send_text(f"Broadcast: {data}")
except Exception as e:
print(f"An error occurred: {e}")
finally:
connected_clients.remove(websocket)

if name == "main":
uvicorn.run(app, host="0.0.0.0", port=8000)
```

2. 路径参数和查询参数

与 HTTP 请求类似，WebSocket 端点也可以接收路径参数和查询参数。

```python
from fastapi import FastAPI, WebSocket
import uvicorn

app = FastAPI()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
await websocket.accept()
query_params = websocket.query_params
print(f"Client ID: {client_id}, Query Params: {query_params}")
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")

if name == "main":
uvicorn.run(app, host="0.0.0.0", port=8000)
```

3. 使用 Depends 进行依赖注入

可以使用 FastAPI 的依赖注入系统，将数据库连接、配置信息等注入到 WebSocket 端点中。

```python
from fastapi import FastAPI, WebSocket, Depends
import uvicorn

app = FastAPI()

async def get_db():
# 模拟数据库连接
db = {"message": "Hello from DB"}
try:
yield db
finally:
# 关闭数据库连接
pass

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, db: dict = Depends(get_db)):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message from DB: {db['message']}")

if name == "main":
uvicorn.run(app, host="0.0.0.0", port=8000)
```

五、总结

本文深入探讨了 FastAPI 中 WebSocket 的使用，从基本概念、使用方法到高级用法，涵盖了构建实时应用所需的关键知识点。通过合理利用 WebSocket 的特性，可以构建出高性能、低延迟、用户体验极佳的 Web 应用。希望本文能帮助你更好地理解和应用 FastAPI WebSocket，打造更强大的实时通信功能。