10.7 与前端 WebSocket 客户端通信

FastAPI WebSocket实战教程：与前端客户端通信详解

FastAPI 教程

本教程详细讲解如何在FastAPI应用中实现WebSocket服务，与前端WebSocket客户端进行实时通信。从环境设置到代码实现，步骤清晰，示例丰富，适合Web开发初学者快速上手。

FastAPI WebSocket教程：与前端客户端通信

引言

WebSocket是一种通信协议，允许客户端和服务器之间建立持久、全双工、低延迟的连接，非常适合实时应用，如聊天室、实时通知或游戏。FastAPI 是一个现代、快速（高性能）的 Web 框架，集成了 WebSocket 支持，让与前端客户端的通信变得简单高效。

本教程将带你一步步学习如何在 FastAPI 中实现 WebSocket，并与前端 WebSocket 客户端进行通信。适合新人学习，内容详细，代码示例清晰。

1. 环境准备

在开始之前，确保你已经安装了 Python 3.7 或更高版本。我们将使用 FastAPI 和 Uvicorn 作为 Web 服务器。

安装依赖

通过 pip 安装 FastAPI、Uvicorn 和 websockets 库（如果需要额外处理 WebSocket）：

pip install fastapi uvicorn websockets

FastAPI：Web 框架。
Uvicorn：ASGI 服务器，用于运行 FastAPI 应用。
websockets：一个可选库，提供更多 WebSocket 工具，但 FastAPI 内置支持 WebSocket，这里为演示安装。

创建项目结构

创建一个新目录，例如 fastapi_websocket_tutorial，并在其中创建一个 Python 文件，如 main.py。

2. 创建 FastAPI 应用

在 main.py 中，首先导入必要的模块并初始化 FastAPI 应用。

from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse

app = FastAPI()  # 创建 FastAPI 应用实例

FastAPI：主类，用于创建应用。
WebSocket：用于处理 WebSocket 连接和消息。
HTMLResponse：可选，用于提供一个简单的 HTML 前端页面。

3. 定义 WebSocket 端点

在 FastAPI 中，WebSocket 端点使用 @app.websocket 装饰器定义。让我们创建一个基本的 WebSocket 路由。

基本 WebSocket 处理

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    # 接受 WebSocket 连接
    await websocket.accept()
    try:
        # 循环处理消息
        while True:
            # 接收文本消息
            data = await websocket.receive_text()
            print(f"Received message: {data}")
            # 发送回复消息
            await websocket.send_text(f"Echo: {data}")
    except Exception as e:
        print(f"WebSocket error: {e}")
    finally:
        # 关闭连接（可选，但推荐）
        await websocket.close()

代码解释：

@app.websocket("/ws")：定义一个 WebSocket 端点，路径为 /ws。
async def websocket_endpoint(websocket: WebSocket)：异步处理函数，websocket 参数是 WebSocket 连接对象。
await websocket.accept()：接受客户端的 WebSocket 连接请求。
循环中，使用 await websocket.receive_text() 接收前端发送的文本消息。
使用 await websocket.send_text() 发送消息回前端，这里示例是简单的回显。
添加异常处理，确保连接安全关闭。

4. 提供前端示例

为了测试 WebSocket，我们可以创建一个简单的前端页面。FastAPI 可以同时服务 WebSocket 和 HTML。

添加一个 GET 路由提供 HTML

在 main.py 中，添加一个路由，返回包含 WebSocket 客户端的 HTML。

@app.get("/")
async def get():
    html_content = """
    <!DOCTYPE html>
    <html>
    <head>
        <title>FastAPI WebSocket 测试</title>
    </head>
    <body>
        <h1>WebSocket 客户端</h1>
        <input type="text" id="messageInput" placeholder="输入消息" />
        <button onclick="sendMessage()">发送</button>
        <div id="messages"></div>
        <script>
            const ws = new WebSocket("ws://localhost:8000/ws");
            ws.onopen = () => {
                console.log("已连接到 WebSocket 服务器！");
                displayMessage("连接成功。");
            };
            ws.onmessage = (event) => {
                console.log("收到服务器消息：", event.data);
                displayMessage("服务器回复：" + event.data);
            };
            ws.onerror = (error) => {
                console.error("WebSocket 错误：", error);
                displayMessage("连接错误。");
            };
            ws.onclose = () => {
                console.log("连接已关闭。");
                displayMessage("连接已断开。");
            };
            function sendMessage() {
                const input = document.getElementById("messageInput");
                const message = input.value;
                if (message) {
                    ws.send(message);
                    displayMessage("你发送了：" + message);
                    input.value = "";
                }
            }
            function displayMessage(text) {
                const messagesDiv = document.getElementById("messages");
                const p = document.createElement("p");
                p.textContent = text;
                messagesDiv.appendChild(p);
            }
        </script>
    </body>
    </html>
    """
    return HTMLResponse(content=html_content)

代码解释：

@app.get("/")：定义根路径的 GET 路由，返回一个 HTML 页面。
HTML 页面包含一个输入框、发送按钮和消息显示区域。
JavaScript 代码连接到 ws://localhost:8000/ws（WebSocket 端点）。
事件处理：
- onopen：连接建立时触发。
- onmessage：收到服务器消息时触发，在页面上显示。
- onerror 和 onclose：处理错误和连接关闭。
sendMessage 函数从输入框获取消息并通过 WebSocket 发送。

5. 运行和测试应用

启动 FastAPI 服务器

在终端中，运行以下命令启动 Uvicorn 服务器：

uvicorn main:app --reload

main:app：指定模块 main 和 FastAPI 应用 app。
--reload：启用热重载，代码修改时自动重启服务器。

服务器启动后，默认运行在 http://localhost:8000。

测试 WebSocket

打开浏览器，访问 http://localhost:8000。
你应该看到 WebSocket 客户端页面。
在输入框中输入消息，点击“发送”按钮。
检查页面上的消息显示和浏览器控制台（按 F12）的输出。
- 你应该看到消息被发送到服务器，并收到回显回复。
- 在服务器终端中，也会打印收到的消息。

示例交互：

用户输入“Hello”，点击发送。
前端显示“你发送了：Hello”。
服务器接收消息并回复“Echo: Hello”。
前端显示“服务器回复：Echo: Hello”。

6. 高级主题和扩展

一旦掌握基础，可以扩展 WebSocket 功能：

多客户端连接管理

如果需要处理多个客户端，可以使用列表存储连接，并广播消息。

from fastapi import WebSocketDisconnect

active_connections = []

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    active_connections.append(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            # 广播消息给所有客户端
            for connection in active_connections:
                await connection.send_text(f"Broadcast: {data}")
    except WebSocketDisconnect:
        active_connections.remove(websocket)

使用列表 active_connections 存储所有 WebSocket 连接。
WebSocketDisconnect 异常处理连接断开，移除列表中的连接。

发送 JSON 数据

WebSocket 支持发送文本或二进制数据。发送 JSON 消息：

import json

# 在服务器端发送 JSON
await websocket.send_text(json.dumps({"message": "Hello", "status": "ok"}))

# 在前端解析 JSON
ws.onmessage = (event) => {
    const data = JSON.parse(event.data);
    console.log(data.message);
};

错误处理和安全性

错误处理：添加更多异常处理，确保应用稳定。
安全性：WebSocket 连接默认不加密；在生产环境中，使用 WSS（WebSocket Secure）通过 HTTPS。在 FastAPI 中，配置 CORS 或使用认证中间件增强安全。

7. 总结和最佳实践

简单上手：FastAPI 的 WebSocket 集成非常直观，只需几行代码即可实现基本通信。
异步支持：WebSocket 在 FastAPI 中是异步的，适合高并发实时应用。
测试工具：使用浏览器开发者工具或 WebSocket 客户端工具（如 wscat）进行测试。
扩展性：根据需求，可以添加连接池、消息队列等来优化性能。

通过本教程，你学会了如何在 FastAPI 中创建 WebSocket 端点，并与前端客户端进行实时通信。继续实践，探索更多 FastAPI 特性，如依赖注入或 WebSocket 路由分组。

进一步学习： 查看 FastAPI 官方文档获取更多高级功能，如 WebSocket 认证或与数据库集成。

提示： 在开发过程中，确保网络环境允许 WebSocket 连接，避免防火墙或代理问题。如果遇到问题，检查浏览器控制台错误和服务器日志。

上一章 10.6 WebSocket 认证与安全

下一章 11.1 小型项目结构（单一文件）