基于 SocketIO 消息协议设计报文规范，构建FastAPI上的SocketIO 应用
<p>最近在研究Python下整合FastAPI的Socket.IO 应用，对于其WebSocket的消息报文协议进行了深入了解，并整理了相关的协议内容，整合到FastAPI的WebSocket通讯处理中，用作多端的消息通讯，如聊天，系统信息通知等。</p><h2 id="1-总体设计">1. 总体设计</h2>
<ul>
<li>统一事件名：客户端和服务端只监听/发送一个事件，例如："message"。</li>
<li>消息类型区分：通过 msgtype 字段区分不同的业务逻辑。</li>
<li>通用字段：每条消息都必须包含的基础字段（如 msgtype, sender, timestamp）。</li>
<li>扩展字段：放在 payload 里，根据不同类型自由定义。</li>
<li>报文格式：统一为 JSON 格式。</li>
<li>广播机制：服务端可以广播消息到所有客户端，方便消息的实时推送。</li>
<li>房间机制：服务端可以创建、加入或离开房间，方便消息的分发。</li>
</ul>
<h2 id="2-消息格式">2. 消息格式</h2>
<pre><code class="language-json">{
"msgtype": "string", // 消息类型，如 "chat", "notification", "command"
"sender": "string", // 发送者标识，如用户ID或系统标识
"timestamp": "number", // 消息发送时间戳（毫秒）
"payload": { // 消息内容（不同 msgtype 有不同格式）
// 根据 msgtype 定义不同的结构
},
"room": "string", // 房间标识（可选），表示消息所属的房间
"touser": "string", // 接收者标识（可选），表示消息的目标用户,多个接收者用‘|’分隔
"totag": "string", // 接收者标签（可选），表示消息的目标用户标签,多个标签用‘|’分隔
}
</code></pre>
<p>例子：</p>
<pre><code class="language-json">{
"msgtype": "chat",
"room": "room1",
"sender": "user1",
"timestamp": 1612345678901,
"payload": {
"text": "Hello, world!"
}
}
</code></pre>
<h2 id="3-消息类型">3. 消息类型</h2>
<h3 id="31-系统类消息system">3.1 系统类消息（system）</h3>
<p>系统类消息由系统发送，用于通知客户端或服务端状态变化。</p>
<pre><code class="language-json">{
"msgtype": "system_notice",
"sender": "system",
"timestamp": 1694412000000,
"payload": {
"level": "info", // info, warning, error
"text": "服务器即将维护"
}
}
</code></pre>
<h3 id="32-聊天类消息chat">3.2 聊天类消息（chat）</h3>
<p>聊天类消息由用户发送，用于聊天。</p>
<pre><code class="language-json">{
"msgtype": "chat_message",
"room": null,
"sender": "alice",
"timestamp": 1694412000000,
"payload": {
"text": "你好，大家！",
"extra": {
"emoji": "😄"
}
}
}
</code></pre>
<h3 id="33-房间操作类消息room">3.3 房间操作类消息（room）</h3>
<p>房间操作类消息用于创建、加入或离开房间。</p>
<pre><code class="language-json">{
"msgtype": "room_event",
"room": "room1",
"sender": "bob",
"timestamp": 1694412000000,
"payload": {
"action": "join", // join, leave, create, destroy
"user": "bob"
}
}
</code></pre>
<h3 id="34-ack--状态反馈消息ack">3.4 ACK / 状态反馈消息（ack）</h3>
<p>ACK / 状态反馈消息用于确认客户端或服务端收到消息。</p>
<pre><code class="language-json">{
"msgtype": "ack",
"sender": "server",
"timestamp": 1694412000000,
"payload": {
"status": "ok", // ok, fail
"requestId": "abc123",
"msg": "消息已送达"
}
}
</code></pre>
<h3 id="35-命令类消息command">3.5 命令类消息（command）</h3>
<p>命令类消息用于向服务端发送指令。</p>
<pre><code class="language-json">{
"msgtype": "command",
"sender": "user1",
"timestamp": 1694412000000,
"payload": {
"command": "create_room",
"args": {
"name": "room1"
}
}
}
</code></pre>
<h3 id="36-图片类消息image">3.6 图片类消息（image）</h3>
<p>图片类消息用于发送图片。</p>
<pre><code class="language-json">{
"msgtype": "image",
"room": "room1",
"sender": "user1",
"timestamp": 1694412000000,
"payload": {
"url": "https://example.com/image.jpg"
}
}
</code></pre>
<h3 id="37-文件类消息file">3.7 文件类消息（file）</h3>
<p>文件类消息用于发送文件。</p>
<pre><code class="language-json">{
"msgtype": "file",
"room": "room1",
"sender": "user1",
"timestamp": 1694412000000,
"payload": {
"url": "https://example.com/file.pdf"
}
}
</code></pre>
<h3 id="38-语音类消息voice">3.8 语音类消息（voice）</h3>
<h3 id="39-视频类消息video">3.9 视频类消息（video）</h3>
<h3 id="310-位置类消息location">3.10 位置类消息（location）</h3>
<h3 id="311-其他类型消息other">3.11 其他类型消息（other）</h3>
<h3 id="312-自定义消息custom">3.12 自定义消息（custom）</h3>
<h2 id="4python-服务端示例">4、Python 服务端示例</h2>
<pre><code class="language-python">import time
import socketio
sio = socketio.AsyncServer(async_mode="asgi", cors_allowed_origins="*")
def make_message(msg_type, sender, payload, room=None):
return {
"msgtype": msg_type,
"room": room,
"sender": sender,
"timestamp": int(time.time() * 1000),
"payload": payload
}
@sio.event
async def chat_message(sid, data):
msg = make_message("chat_message", sid, {"text": data})
await sio.emit("message", msg)# 广播
@sio.event
async def join_room(sid, room):
sio.enter_room(sid, room)
msg = make_message("room_event", "system", {"action": "join", "user": sid}, room)
await sio.emit("message", msg, room=room)
</code></pre>
<h2 id="5vue3-前端示例">5、Vue3 前端示例</h2>
<pre><code class="language-javascript">socket.on("message", (msg) => {
switch (msg.msgtype) {
case "chat_message":
console.log(`[${msg.sender}] 说: ${msg.payload.text}`);
break;
case "system_notice":
console.log(`系统通知: ${msg.payload.text}`);
break;
case "room_event":
console.log(`房间 ${msg.room} 事件: ${msg.payload.action} -> ${msg.payload.user}`);
break;
case "ack":
console.log(`ACK: ${msg.payload.status} (${msg.payload.msg})`);
break;
default:
console.warn("未知消息类型:", msg);
}
});
</code></pre>
<h2 id="6通信流程示例">6、通信流程示例</h2>
<h3 id="61-普通消息">6.1 普通消息</h3>
<ul>
<li>客户端发送：</li>
</ul>
<pre><code class="language-json">{
"msgtype": "chat_message",
"sender": "alice",
"timestamp": 1694412000000,
"payload": {
"text": "Hello World"
}
}
</code></pre>
<ul>
<li>服务端广播：</li>
</ul>
<pre><code class="language-json">{
"msgtype": "chat_message",
"sender": "alice",
"timestamp": 1694412000500,
"payload": {
"text": "Hello World"
}
}
</code></pre>
<h3 id="62-房间消息">6.2 房间消息</h3>
<h4 id="621-创建房间">6.2.1 创建房间</h4>
<ol>
<li>客户端发送：</li>
</ol>
<pre><code class="language-json">{
"msgtype": "command",
"sender": "bob",
"timestamp": 1694412000000,
"payload": {
"command": "create_room",
"args": {
"name": "room1"
}
}
}
2. 服务端响应：
```json
{
"msgtype": "ack",
"sender": "server",
"timestamp": 1694412000000,
"payload": {
"status": "ok",
"requestId": "abc123",
"msg": "房间 room1 创建成功"
}
}
</code></pre>
<h4 id="622-加入房间">6.2.2 加入房间</h4>
<p>1.客户端发送：</p>
<pre><code class="language-json">{
"msgtype": "room_event",
"sender": "alice",
"timestamp": 1694412000000,
"payload": {
"action": "join",
"user": "alice"
}
}
</code></pre>
<p>2.服务端响应：</p>
<pre><code class="language-json">{
"msgtype": "ack",
"sender": "server",
"timestamp": 1694412000000,
"payload": {
"status": "ok",
"requestId": "abc123",
"msg": "用户 alice 加入房间 room1"
}
}
</code></pre>
<p>3.服务端广播：</p>
<pre><code class="language-json">{
"msgtype": "room_event",
"room": "room1",
"sender": "server",
"timestamp": 1694412000000,
"payload": {
"action": "join",
"user": "alice"
}
}
</code></pre>
<h4 id="623-离开房间">6.2.3 离开房间</h4>
<p>1.客户端发送：</p>
<pre><code class="language-json">{
"msgtype": "room_event",
"sender": "alice",
"timestamp": 1694412000000,
"payload": {
"action": "leave",
"user": "alice"
}
}
</code></pre>
<p>2.服务端响应：</p>
<pre><code class="language-json">{
"msgtype": "ack",
"sender": "server",
"timestamp": 1694412000000,
"payload": {
"status": "ok",
"requestId": "abc123",
"msg": "用户 alice 离开房间 room1"
}
}
</code></pre>
<h2 id="7优点">7、优点</h2>
<ul>
<li>
<p>统一事件名：客户端和服务端只监听/发送一个事件，降低耦合度，提高代码可读性。</p>
</li>
<li>
<p>消息类型区分：通过 msgtype 字段区分不同的业务逻辑，提高消息处理效率。</p>
</li>
<li>
<p>通用字段：每条消息都必须包含的基础字段（如 msgtype, sender, timestamp），方便服务端处理。</p>
</li>
<li>
<p>扩展字段：放在 payload 里，根据不同类型自由定义，方便消息内容的扩展。</p>
</li>
<li>
<p>状态反馈消息：ACK / 状态反馈消息用于确认客户端或服务端收到消息，方便客户端处理。</p>
</li>
<li>
<p>命令类消息：命令类消息用于向服务端发送指令，方便服务端处理。</p>
</li>
<li>
<p>广播机制：服务端可以广播消息到所有客户端，方便消息的实时推送。</p>
</li>
<li>
<p>房间机制：服务端可以创建、加入或离开房间，方便消息的分发。</p>
</li>
<li>
<p>扩展性：服务端和客户端可以自由扩展消息类型，增加更多的业务逻辑。</p>
</li>
<li>
<p>兼容性：服务端和客户端可以兼容 SocketIO 协议，方便与其他服务端和客户端通信。</p>
</li>
<li>
<p>易用性：服务端和客户端可以直接使用 SocketIO 库，降低学习成本。</p>
</li>
<li>
<p>性能：服务端和客户端可以利用 WebSocket 协议，提高消息传输效率。</p>
</li>
</ul>
<h2 id="8关于-socketio的挂载的注意事项">8、关于 Socket.IO的挂载的注意事项</h2>
<p>Socket.IO 是一个基于 WebSocket 的实时通信协议，它可以实现浏览器和服务器之间的双向通信。<br>
Socket.IO 协议的实现主要依赖于两个部分：</p>
<ul>
<li>Socket.IO 服务器：Socket.IO 服务器负责维护连接，处理消息，并向客户端推送消息。</li>
<li>Socket.IO 客户端：Socket.IO 客户端负责维护连接，向 Socket.IO 服务器发送消息，并接收消息。</li>
</ul>
<p>如果你用 app = register_app()，然后用 uvicorn main:app 或者 config = uvicorn.Config(app, ...)，这样启动的“app”是原生 FastAPI 实例，不包含 Socket.IO 的功能，所以前端 Socket.IO 客户端连接时不会触发 @sio.event 的回调！</p>
<p><strong>解决方法：</strong><br>
你需要让 Uvicorn 启动的是 ASGI 的组合应用，即 sio_app，而不是纯 FastAPI 的 app。</p>
<pre><code class="language-python">
import socketio
from fastapi import FastAPI
def register_app():
app = FastAPI()
# ...注册路由等...
return app
app = register_app()
sio = socketio.AsyncServer(async_mode='asgi', cors_allowed_origins='*')
sio_app = socketio.ASGIApp(sio, other_asgi_app=app)
# main.py 入口
if __name__ == "__main__":
import uvicorn
# 下面两种写法都可以
uvicorn.run(sio_app, host="0.0.0.0", port=8000)
# 或
# config = uvicorn.Config(sio_app, host="0.0.0.0", port=8000)
# server = uvicorn.Server(config)
# server.run()
</code></pre>
<p>注意：一定要启动的是 sio_app，不是 app！<br>
你可以继续用 register_app() 组织你的 FastAPI 配置，但最终给 Uvicorn 的必须是 sio_app。</p>
<p>最后我们来看看再FastAPI项目上对接的例子通讯截图（尚未完工，持续改进优化中）。<br><br><br>
<strong>服务端的信息截图</strong><br>
<img src="https://img2024.cnblogs.com/blog/8867/202509/8867-20250919102703684-1518866115.png" alt="image" loading="lazy"><br>
<strong>客户端信息截图</strong><br>
<img src="https://img2024.cnblogs.com/blog/8867/202509/8867-20250919102653726-864795853.png" alt="image" loading="lazy"></p>
</div>
<div id="MySignature" role="contentinfo">
<div style="border-right-color: #cccccc; border-right-width: 1px; border-right-style: solid; padding-right: 5px; border-top-color: #cccccc; border-top-width: 1px; border-top-style: solid; padding-left: 4px; font-size: 13px; padding-bottom: 4px; border-left-color: #cccccc; border-left-width: 1px; border-left-style: solid; width: 98%; padding-top: 4px; border-bottom-color: #cccccc; border-bottom-width: 1px; border-bottom-style: solid; background-color: #eeeeee;">
<img src="http://www.cnblogs.com/Images/OutliningIndicators/None.gif" align="top" alt>
<span style="color: #000000"><span class="Apple-tab-span" style="white-space: pre"></span>
专注于代码生成工具、.Net/Python 框架架构及软件开发，以及各种Vue.js的前端技术应用。著有Winform开发框架/混合式开发框架、微信开发框架、Bootstrap开发框架、ABP开发框架、SqlSugar开发框架、Python开发框架等框架产品。
<br> 转载请注明出处：撰写人：伍华聪 http://www.iqidi.com <br> </span></div><br><br>
来源：https://www.cnblogs.com/wuhuacong/p/19100261
頁:
[1]