登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 业界 ›FastAPI WebSocket：你的双向通信通道为何如此丝滑？ ...

FastAPI WebSocket：你的双向通信通道为何如此丝滑？

毡轩 9 小时前
1.jpeg
2.jpeg
扫描二维码
关注或者微信搜一搜：编程智域 前端至全栈交流与成长
发现1000+提升效率与开发的AI工具和实用程序：https://tools.cmdragon.cn/
  1. # 示例代码运行环境
  2. # Python 3.8+
  3. # 安装依赖：pip install fastapi==0.68.0 uvicorn==0.15.0 websockets==10.3 pydantic==1.10.7
复制代码
一、WebSocket路由声明规范

1.1 基础路由定义

使用@app.websocket装饰器声明WebSocket路由：
  1. from fastapi import FastAPI, WebSocket
  3. app = FastAPI()
  6. @app.websocket("/ws/chat/{room_id}")
  7. async def websocket_chat(websocket: WebSocket, room_id: int):
  8.     await websocket.accept()
  9.     try:
  10.         while True:
  11.             data = await websocket.receive_text()
  12.             await websocket.send_text(f"Room {room_id}: {data}")
  13.     except WebSocketDisconnect:
  14.         print("Client disconnected")
复制代码
关键要点解析：

  • 路径参数通过URL直接传递（如{room_id}）
  • WebSocket对象自动注入到路由函数
  • 必须显式调用await websocket.accept()建立连接
  • receive_text()和send_text()实现双向通信
1.2 路由参数验证

结合路径参数与查询参数进行复合验证：
  1. from fastapi import Query, WebSocket
  4. @app.websocket("/ws/secure/{client_id}")
  5. async def secure_ws(
  6.         websocket: WebSocket,
  7.         client_id: int = Path(..., gt=0),
  8.         token: str = Query(..., min_length=8)
  9. ):
  10.     if not validate_token(token):
  11.         await websocket.close(code=1008)
  12.         return
  13.     # ...连接处理逻辑...
复制代码
二、客户端连接建立与握手验证

2.1 握手过程控制

自定义握手验证流程：
  1. @app.websocket("/ws/auth")
  2. async def auth_ws(websocket: WebSocket):
  3.     # 手动控制握手过程
  4.     await websocket.accept()
  6.     # 获取握手时的请求头
  7.     headers = websocket.headers
  8.     auth_token = headers.get("authorization")
  10.     if not verify_jwt_token(auth_token):
  11.         await websocket.close(code=1008, reason="Invalid credentials")
  12.         return
  14.     # 验证通过后的处理逻辑
复制代码
2.2 验证失败处理模式

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

3.1 连接状态跟踪

使用字典维护活跃连接：
  1. from typing import Dict
  3. active_connections: Dict[str, WebSocket] = {}
  6. @app.websocket("/ws/status")
  7. async def status_ws(websocket: WebSocket):
  8.     await websocket.accept()
  9.     client_id = str(websocket.client)
  10.     active_connections[client_id] = websocket
  12.     try:
  13.         while True:
  14.             # 保持连接活跃
  15.             await websocket.receive_text()
  16.     except WebSocketDisconnect:
  17.         del active_connections[client_id]
复制代码
3.2 心跳检测机制
  1. import asyncio
  4. async def heartbeat(websocket: WebSocket, interval: int = 30):
  5.     try:
  6.         while True:
  7.             await asyncio.sleep(interval)
  8.             await websocket.send_json({
  9.                 "type": "heartbeat",
  10.                 "timestamp": time.time()
  11.             })
  12.     except WebSocketDisconnect:
  13.         print("Heartbeat terminated")
  16. @app.websocket("/ws/realtime")
  17. async def realtime_ws(websocket: WebSocket):
  18.     await websocket.accept()
  19.     # 启动独立心跳任务
  20.     heartbeat_task = asyncio.create_task(heartbeat(websocket))
  22.     try:
  23.         # 主消息处理循环
  24.         while True:
  25.             data = await websocket.receive_json()
  26.             # 处理业务逻辑...
  27.     finally:
  28.         heartbeat_task.cancel()
  29.         await websocket.close()
复制代码
课后 Quiz


  • 当客户端发送非文本格式数据时如何处理？
  1. # 正确处理方法
  2. try:
  3.     data = await websocket.receive_json()
  4. except WebSocketDisconnect:
  5. # 处理断开逻辑
  6. except ValueError:
  7.     await websocket.send_text("ERROR: 仅支持JSON格式")
  9. # 错误处理方法：直接使用未校验的receive()
复制代码

  • 如何获取客户端的真实IP地址？
  1. # 正确方法
  2. client_ip = websocket.client.host
  4. # 常见错误：直接读取X-Forwarded-For头
  5. # 需要配合代理设置处理
复制代码
常见报错解决方案

错误1：403 Handshake Failed

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

  • 缺少必要认证头
  • 验证逻辑返回False后未及时close()
    解决：
  1. # 在拒绝连接时立即关闭
  2. if not valid:
  3.     await websocket.close(code=1008)
  4.     return  # 必须立即返回
复制代码
错误2：1006 Abnormal Closure

现象：客户端非正常断开
处理方案：
  1. try:
  2.     while True:
  3.         data = await websocket.receive_text()
  4. except WebSocketDisconnect as e:
  5.     print(f"断开代码：{e.code}")
复制代码
错误3：TypeError: 'str' expected

场景：发送非文本数据时
正确做法：
  1. # 发送二进制数据
  2. await websocket.send_bytes(binary_data)
  4. # 发送文本数据
  5. await websocket.send_text("message")
  7. # 发送JSON
  8. 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文本生成图像 - 应用商店 | 免费好用的在线工具
  • 临时邮箱 - 应用商店 | 免费好用的在线工具
  • 二维码解析器 - 应用商店 | 免费好用的在线工具
  • 文本转思维导图 - 应用商店 | 免费好用的在线工具
  • 正则表达式可视化工具 - 应用商店 | 免费好用的在线工具
  • 文件隐写工具 - 应用商店 | 免费好用的在线工具
  • IPTV 频道探索器 - 应用商店 | 免费好用的在线工具
  • 快传 - 应用商店 | 免费好用的在线工具
  • 随机抽奖工具 - 应用商店 | 免费好用的在线工具
  • 动漫场景查找器 - 应用商店 | 免费好用的在线工具
  • 时间工具箱 - 应用商店 | 免费好用的在线工具
  • 网速测试 - 应用商店 | 免费好用的在线工具
  • AI 智能抠图工具 - 应用商店 | 免费好用的在线工具
  • 背景替换工具 - 应用商店 | 免费好用的在线工具
  • 艺术二维码生成器 - 应用商店 | 免费好用的在线工具
  • Open Graph 元标签生成器 - 应用商店 | 免费好用的在线工具
  • 图像对比工具 - 应用商店 | 免费好用的在线工具
  • 图片压缩专业版 - 应用商店 | 免费好用的在线工具
  • 密码生成器 - 应用商店 | 免费好用的在线工具
  • SVG优化器 - 应用商店 | 免费好用的在线工具
  • 调色板生成器 - 应用商店 | 免费好用的在线工具
  • 在线节拍器 - 应用商店 | 免费好用的在线工具
  • IP归属地查询 - 应用商店 | 免费好用的在线工具
  • CSS网格布局生成器 - 应用商店 | 免费好用的在线工具
  • 邮箱验证工具 - 应用商店 | 免费好用的在线工具
  • 书法练习字帖 - 应用商店 | 免费好用的在线工具
  • 金融计算器套件 - 应用商店 | 免费好用的在线工具
  • 中国亲戚关系计算器 - 应用商店 | 免费好用的在线工具
  • Protocol Buffer 工具箱 - 应用商店 | 免费好用的在线工具
  • IP归属地查询 - 应用商店 | 免费好用的在线工具
  • 图片无损放大 - 应用商店 | 免费好用的在线工具
  • 文本比较工具 - 应用商店 | 免费好用的在线工具
  • IP批量查询工具 - 应用商店 | 免费好用的在线工具
  • 域名查询工具 - 应用商店 | 免费好用的在线工具
  • DNS工具箱 - 应用商店 | 免费好用的在线工具
  • 网站图标生成器 - 应用商店 | 免费好用的在线工具
  • XML Sitemap

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
毡轩
9 小时前

0

粉丝关注

18

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫