消息帧结构

本文档引用的文件

引言

本文档详细描述了基于WebSocket协议的消息帧结构设计与实现，重点分析了消息的二进制/文本帧格式、头部信息、负载数据结构和长度编码方式。文档基于``项目中的WebSocket相关组件，深入解析了websocket.constants.ts中定义的帧类型常量及其在实际通信中的应用。

系统采用基于主题（topic）的发布-订阅模式，实现了客户端与服务端之间的实时双向通信。通过分析服务端如何解析和验证消息帧的完整性，包括校验和计算、长度检查和边界处理，本文档提供了完整的消息帧字节级布局图示，展示了各字段的位置和大小。

项目结构分析

项目采用分层架构设计，WebSocket相关功能主要分布在deno/lib/websocket/目录下，包含常量定义、路由处理和数据访问对象（DAO）三个核心组件。前端Vue组件位于pc/src/compositions/websocket.ts，负责客户端连接管理和消息收发。

图表来源

本节来源

核心数据结构与常量

全局状态映射

系统定义了三个核心的Map数据结构，用于维护WebSocket连接的状态：

typescript

// deno-lint-ignore no-explicit-any
export const callbacksMap = new Map<string, ((data: any) => Promise<void> | void)[]>();

export const socketMap = new Map<string, WebSocket[]>();

export const clientIdTopicsMap = new Map<string, string[]>();

字段说明：

callbacksMap: 主题回调映射，键为topic字符串，值为该主题订阅的回调函数数组
socketMap: 客户端连接映射，键为clientId字符串，值为该客户端的所有WebSocket连接数组
clientIdTopicsMap: 客户端主题映射，键为clientId字符串，值为该客户端订阅的主题字符串数组

这些Map结构实现了O(1)时间复杂度的查找操作，确保了高并发场景下的消息分发效率。

本节来源

websocket.constants.ts

WebSocket消息协议设计

消息类型定义

系统支持三种主要的消息操作类型：

操作类型	描述	方向
subscribe	订阅主题	客户端 → 服务端
publish	发布消息	客户端 ↔ 服务端
unSubscribe	取消订阅	客户端 → 服务端

消息帧格式

消息帧采用JSON格式进行序列化，具有统一的结构：

json

{
  "action": "subscribe",
  "data": {
    "topics": ["topic1", "topic2"]
  }
}

或

json

{
  "action": "publish",
  "data": {
    "topic": "user.update",
    "payload": { "userId": "123", "name": "张三" }
  }
}

字段说明：

action: 消息操作类型，枚举值为"subscribe"、"publish"、"unSubscribe"
data: 消息数据体，根据action类型具有不同的结构

心跳机制

系统实现了ping-pong心跳机制，用于保持连接活跃：

typescript

function socketPing() {
  if (socket && socket.readyState === WebSocket.OPEN) {
    socket.send("ping");
  }
  setTimeout(socketPing, 60000);
}

心跳间隔为60秒，服务端收到"ping"消息后立即回复"pong"。

本节来源

服务端消息处理流程

连接升级流程

服务端通过/api/websocket/upgrade端点处理WebSocket连接升级请求：

图表来源

websocket.router.ts

本节来源

websocket.router.ts

消息分发流程

服务端收到消息后，根据action类型进行分发处理：

图表来源

websocket.router.ts

本节来源

websocket.router.ts

客户端连接与通信机制

连接管理

客户端实现了自动重连机制，确保网络不稳定时的连接可靠性：

typescript

async function reConnect() {
  if (socket && (socket.readyState === WebSocket.OPEN || socket.readyState === WebSocket.CONNECTING)) {
    return socket;
  }
  if (topicCallbackMap.size === 0) {
    return;
  }
  reConnectNum++;
  let time = 200;
  if (reConnectNum > 10) {
    time = 5000;
  } else {
    time = reConnectNum * 200;
  }
  await new Promise((resolve) => setTimeout(resolve, time));
  await connect();
}

重连策略：

初始重连间隔：200ms
递增策略：每次增加200ms
最大重连间隔：5000ms（当重连次数超过10次时）

消息订阅机制

客户端通过subscribe函数实现主题订阅：

typescript

export async function subscribe<T>(
  topic: string,
  callback: ((data: T | undefined) => void),
) {
  const socket = await connect();
  if (!socket) {
    return;
  }
  let callbacks = topicCallbackMap.get(topic);
  if (!callbacks) {
    callbacks = [ ];
    topicCallbackMap.set(topic, callbacks);
  }
  callbacks.push(callback);
  socket.send(JSON.stringify({
    action: "subscribe",
    data: {
      topics: [ topic ],
    },
  }));
}

本节来源

websocket.ts