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

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 业界 ›如何在FastAPI中玩转WebSocket消息处理？

如何在FastAPI中玩转WebSocket消息处理？

焦尔蕾 11 小时前
1.jpeg
2.jpeg
扫描二维码
关注或者微信搜一搜：编程智域 前端至全栈交流与成长
发现1000+提升效率与开发的AI工具和实用程序：https://tools.cmdragon.cn/
一、文本消息接收与发送
  1. # 运行环境：Python 3.8+
  2. # 安装依赖：pip install fastapi==0.68.0 uvicorn==0.15.0 websockets==10.3 pydantic==1.10.7
  4. from fastapi import FastAPI, WebSocket
  6. app = FastAPI()
  9. @app.websocket("/ws/chat")
  10. async def websocket_chat(websocket: WebSocket):
  11.     await websocket.accept()
  12.     try:
  13.         while True:
  14.             # 接收文本消息
  15.             client_msg = await websocket.receive_text()
  16.             # 处理消息（示例：添加时间戳）
  17.             server_response = f"[{datetime.now()}] Server received: {client_msg}"
  18.             # 发送文本响应
  19.             await websocket.send_text(server_response)
  20.     except WebSocketDisconnect:
  21.         print("Client disconnected")
复制代码
消息处理流程：


  • 客户端通过ws://协议建立WebSocket连接
  • 服务端使用await websocket.accept()接受连接
  • 进入接收循环处理receive_text()和send_text()
  • 异常处理自动断开连接
sequenceDiagram    participant C as Client    participant S as Server        C->>S: Establish WebSocket connection (ws://)    activate S    S->>S: await websocket.accept() to accept connection    activate C    loop Receiving and Sending Messages        C->>S: receive_text()        S-->>C: send_text()    end    alt Exception Handling        S->>C: Automatically disconnect    end    deactivate S    deactivate C应用场景：实时聊天室、协同编辑系统、实时日志监控
二、二进制数据传输处理
  1. @app.websocket("/ws/file-transfer")
  2. async def websocket_file(websocket: WebSocket):
  3.     await websocket.accept()
  4.     try:
  5.         while True:
  6.             # 接收二进制数据
  7.             binary_data = await websocket.receive_bytes()
  8.             # 保存文件示例
  9.             with open("received_file.bin", "wb") as f:
  10.                 f.write(binary_data)
  11.             # 发送确认消息
  12.             await websocket.send_bytes(b"FILE_RECEIVED")
  13.     except WebSocketDisconnect:
  14.         print("File transfer interrupted")
复制代码
二进制处理要点：


  • 使用receive_bytes()和send_bytes()方法
  • 适合传输图片、音频、视频等二进制格式
  • 建议分块传输大文件（结合消息头协议）
三、JSON消息序列化与自动解析
  1. from pydantic import BaseModel
  4. class MessageModel(BaseModel):
  5.     user: str
  6.     content: str
  7.     timestamp: float
  10. @app.websocket("/ws/json-demo")
  11. async def websocket_json(websocket: WebSocket):
  12.     await websocket.accept()
  13.     try:
  14.         while True:
  15.             json_data = await websocket.receive_json()
  16.             # 自动验证JSON结构
  17.             message = MessageModel(**json_data)
  18.             # 处理业务逻辑
  19.             processed_data = message.dict()
  20.             processed_data["status"] = "PROCESSED"
  21.             # 返回处理结果
  22.             await websocket.send_json(processed_data)
  23.     except ValidationError as e:
  24.         await websocket.send_json({"error": str(e)})
复制代码
自动验证流程：


  • 接收原始JSON数据
  • 通过Pydantic模型进行数据清洗
  • 自动类型转换和字段验证
  • 结构化错误响应返回
四、消息接收循环与超时控制
  1. from websockets.exceptions import ConnectionClosed
  4. @app.websocket("/ws/with-timeout")
  5. async def websocket_timeout(websocket: WebSocket):
  6.     await websocket.accept()
  7.     try:
  8.         while True:
  9.             try:
  10.                 # 设置10秒接收超时
  11.                 data = await asyncio.wait_for(websocket.receive_text(), timeout=10)
  12.                 await process_message(data)
  13.             except asyncio.TimeoutError:
  14.                 # 发送心跳包保持连接
  15.                 await websocket.send_text("HEARTBEAT")
  16.     except ConnectionClosed:
  17.         print("Connection closed normally")
复制代码
超时控制策略：


  • 使用asyncio.wait_for设置单次接收超时
  • 定期发送心跳包维持连接
  • 异常分类处理（正常关闭/异常断开）
课后Quiz

Q1：如何处理同时接收文本和二进制消息的场景？
A：通过receive()方法获取消息类型判断：
  1. message = await websocket.receive()
  2. if message["type"] == "websocket.receive.text":
  3.     handle_text(message["text"])
  4. elif message["type"] == "websocket.receive.bytes":
  5.     handle_bytes(message["bytes"])
复制代码
Q2：为什么推荐使用Pydantic进行JSON验证？
A：① 自动类型转换 ② 字段约束检查 ③ 防御无效数据 ④ 生成API文档
常见报错解决方案

422 Validation Error
  1. {
  2.   "detail": [
  3.     {
  4.       "loc": [
  5.         "body",
  6.         "timestamp"
  7.       ],
  8.       "msg": "field required",
  9.       "type": "value_error.missing"
  10.     }
  11.   ]
  12. }
复制代码
解决方法：

  • 检查客户端发送的JSON字段是否完整
  • 验证时间戳是否为数字类型
  • 添加默认值字段：timestamp: float = None
WebSocketTimeoutException

  • 成因：长时间未发送/接收消息
  • 解决：调整wait_for超时参数，添加心跳机制
余下文章内容请点击跳转至 个人博客页面 或者 扫码关注或者微信搜一搜：编程智域 前端至全栈交流与成长
，阅读完整的文章：如何在FastAPI中玩转WebSocket消息处理？
往期文章归档：


  • 如何在FastAPI中玩转WebSocket，让实时通信不再烦恼？ - cmdragon's Blog
  • 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
免费好用的热门在线工具


  • 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
您需要登录后才可以回帖 登录 | 立即注册
发帖
焦尔蕾
11 小时前

0

粉丝关注

11

主题发布

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