文章标题：

FastAPI中WebSocket的运用：探索高效双向交互之道

文章内容：

示例代码运行环境

Python 3.8+

安装依赖：pip install fastapi==0.68.0 uvicorn==0.15.0 websockets==10.3 pydantic==1.10.7

一、WebSocket路由声明规范

1.1 基础路由定义

通过@app.websocket装饰器来定义WebSocket路由：

from fastapi import FastAPI, WebSocket

app = FastAPI()


@app.websocket("/ws/chat/{room_id}")
async def websocket_chat(websocket: WebSocket, room_id: int):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Room {room_id}: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")

关键要点解读：
* 路径参数通过URL直接传递（如{room_id}）
* WebSocket对象会自动注入到路由函数中
* 必须显式调用await websocket.accept()来建立连接
* 通过receive_text()和send_text()实现双向的数据通信

1.2 路由参数验证

结合路径参数与查询参数进行复合验证：

from fastapi import Query, WebSocket


@app.websocket("/ws/secure/{client_id}")
async def secure_ws(
        websocket: WebSocket,
        client_id: int = Path(..., gt=0),
        token: str = Query(..., min_length=8)
):
    if not validate_token(token):
        await websocket.close(code=1008)
        return
    # ...连接处理逻辑...


## 二、客户端连接建立与握手验证

### 2.1 握手过程控制
自定义握手验证流程：

```python
@app.websocket("/ws/auth")
async def auth_ws(websocket: WebSocket):
    # 手动控制握手过程
    await websocket.accept()

    # 获取握手时的请求头
    headers = websocket.headers
    auth_token = headers.get("authorization")

    if not verify_jwt_token(auth_token):
        await websocket.close(code=1008, reason="Invalid credentials")
        return

    # 验证通过后的处理逻辑

2.2 验证失败处理模式

graph TD
A[客户端请求连接] --> B{验证Headers}
B -- 成功 --> C[返回101状态码]
B -- 失败 --> D[返回403状态码]
C --> E[建立双向通信]
D --> F[关闭TCP连接]

三、连接状态维护与生命周期管理

3.1 连接状态跟踪

使用字典来维护活跃的连接：

from typing import Dict

active_connections: Dict[str, WebSocket] = {}


@app.websocket("/ws/status")
async def status_ws(websocket: WebSocket):
    await websocket.accept()
    client_id = str(websocket.client)
    active_connections[client_id] = websocket

    try:
        while True:
            # 保持连接活跃
            await websocket.receive_text()
    except WebSocketDisconnect:
        del active_connections[client_id]

3.2 心跳检测机制

import asyncio


async def heartbeat(websocket: WebSocket, interval: int = 30):
    try:
        while True:
            await asyncio.sleep(interval)
            await websocket.send_json({
                "type": "heartbeat",
                "timestamp": time.time()
            })
    except WebSocketDisconnect:
        print("Heartbeat terminated")


@app.websocket("/ws/realtime")
async def realtime_ws(websocket: WebSocket):
    await websocket.accept()
    # 启动独立心跳任务
    heartbeat_task = asyncio.create_task(heartbeat(websocket))

    try:
        # 主消息处理循环
        while True:
            data = await websocket.receive_json()
            # 处理业务逻辑...
    finally:
        heartbeat_task.cancel()
        await websocket.close()


## 课后 Quiz

  1. 当客户端发送非文本格式数据时如何处理？

```python
# 正确处理方法
try:
    data = await websocket.receive_json()
except WebSocketDisconnect:
    # 处理断开逻辑
except ValueError:
    await websocket.send_text("ERROR: 仅支持JSON格式")

# 错误处理方法：直接使用未校验的receive()

1. 如何获取客户端的真实IP地址？

# 正确方法
client_ip = websocket.client.host

# 常见错误：直接读取X-Forwarded-For头
# 需要配合代理设置处理

常见报错解决方案

错误1：403 Handshake Failed

现象 ：客户端连接时立即断开
分析 ：未通过握手验证，常见于：

• 缺少必要认证头
• 验证逻辑返回False后未及时close()
  解决 ：

# 在拒绝连接时立即关闭
if not valid:
    await websocket.close(code=1008)
    return  # 必须立即返回

错误2：1006 Abnormal Closure

现象 ：客户端非正常断开
处理方案 ：

try:
    while True:
        data = await websocket.receive_text()
except WebSocketDisconnect as e:
    print(f"断开代码：{e.code}")

错误3：TypeError: 'str' expected

场景 ：发送非文本数据时
正确做法 ：

# 发送二进制数据
await websocket.send_bytes(binary_data)

# 发送文本数据
await websocket.send_text("message")

# 发送JSON
await websocket.send_json({"key": "value"})

余下文章内容请点击跳转至 个人博客页面 或者 扫码关注或者微信搜一搜：编程智域 前端至全栈交流与成长
，阅读完整的文章：如何在FastAPI中玩转WebSocket，让实时通信不再烦恼？

往期文章归档：

• WebSocket与HTTP协议究竟有何不同？FastAPI如何让长连接变得如此简单？ - cmdragon's Blog
• FastAPI如何玩转安全防护，让黑客望而却步？
• 如何用三层防护体系打造坚不可摧的 API 安全堡垒？ - cmdragon's Blog
• FastAPI安全加固：密钥轮换、限流策略与安全头部如何实现三重防护？ - cmdragon's Blog
• 如何在FastAPI中巧妙玩转数据脱敏，让敏感信息安全无忧？ - cmdragon's Blog
• RBAC权限模型如何让API访问控制既安全又灵活？ - cmdragon's Blog
• FastAPI中的敏感数据如何在不泄露的情况下翩翩起舞？
• FastAPI安全认证的终极秘籍：OAuth2与JWT如何完美融合？ - cmdragon's Blog
• 如何在FastAPI中打造坚不可摧的Web安全防线？ - cmdragon's Blog
• 如何用 FastAPI 和 RBAC 打造坚不可摧的安全堡垒？ - cmdragon's Blog
• FastAPI权限配置：你的系统真的安全吗？ - cmdragon's Blog
• FastAPI权限缓存：你的性能瓶颈是否藏在这只“看不见的手”里？ | cmdragon's Blog
• FastAPI日志审计：你的权限系统是否真的安全无虞？ | cmdragon's Blog
• 如何在FastAPI中打造坚不可摧的安全防线？ | cmdragon's Blog
• 如何在FastAPI中实现权限隔离并让用户乖乖听话？ | cmdragon's Blog
• 如何在FastAPI中玩转权限控制与测试，让代码安全又优雅？ | cmdragon's Blog
• 如何在FastAPI中打造一个既安全又灵活的权限管理系统？ | cmdragon's Blog
• FastAPI访问令牌的权限声明与作用域管理：你的API安全真的无懈可击吗？ | cmdragon's Blog
• 如何在FastAPI中构建一个既安全又灵活的多层级权限系统？ | cmdragon's Blog
• FastAPI如何用角色权限让Web应用安全又灵活？ | cmdragon's Blog
• FastAPI权限验证依赖项究竟藏着什么秘密？ | cmdragon's Blog
• 如何用FastAPI和Tortoise-ORM打造一个既高效又灵活的角色管理系统？ | cmdragon's Blog
• JWT令牌如何在FastAPI中实现安全又高效的生成与验证？ | cmdragon's Blog
• 你的密码存储方式是否在向黑客招手？ | cmdragon's Blog
• 如何在FastAPI中轻松实现OAuth2认证并保护你的API？ | cmdragon's Blog
• FastAPI安全机制：从OAuth2到JWT的魔法通关秘籍 | cmdragon's Blog
• FastAPI认证系统：从零到令牌大师的奇幻之旅 | cmdragon's Blog
• FastAPI安全异常处理：从401到422的奇妙冒险 | cmdragon's Blog
• FastAPI权限迷宫：RBAC与多层级依赖的魔法通关秘籍 | cmdragon's Blog
• JWT令牌：从身份证到代码防伪的奇妙之旅 | cmdragon's Blog
• FastAPI安全认证：从密码到令牌的魔法之旅 | cmdragon's Blog
• 密码哈希：Bcrypt的魔法与盐值的秘密 | cmdragon's Blog
• 用户认证的魔法配方：从模型设计到密码安全的奇幻之旅 | cmdragon's Blog
• FastAPI安全门神：OAuth2PasswordBearer的奇妙冒险 | cmdragon's Blog
• OAuth2密码模式：信任的甜蜜陷阱与安全指南 | cmdragon's Blog
• API安全大揭秘：认证与授权的双面舞会 | cmdragon's Blog

免费好用的热门在线工具

• CMDragon 在线工具 - 高级AI工具箱与开发者套件 | 免费好用的在线工具
• 应用商店 - 发现1000+提升效率与开发的AI工具和实用程序 | 免费好用的在线工具
• CMDragon 更新日志 - 最新更新、功能与改进 | 免费好用的在线工具
• 支持我们 - 成为赞助者 | 免费好用的在线工具
• AI文本生成图像 - 应用商店 | 免费好用的在线工具
• [临时邮箱 - 应用