WebSocket 连接建立与丢包处理机制详解-CFANZ编程社区

连接建立：WebSocket 握手过程

1. 客户端发起 HTTP 升级请求

GET /chat HTTP/1.1
Host: server.example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Version: 13
Origin: http://example.com

关键字段:

Upgrade: websocket - 请求协议升级
Connection: Upgrade - 连接升级
Sec-WebSocket-Key - 随机 Base64 编码密钥
Sec-WebSocket-Version - 协议版本

2. 服务端响应协议升级

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=

验证原理:

function generateAcceptKey(clientKey) {
  const crypto = require('crypto');
  const GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11';
  return crypto.createHash('sha1')
              .update(clientKey + GUID)
              .digest('base64');
}

3. 连接建立完成

TCP连接保持打开状态
通信协议切换到WebSocket二进制帧格式

WebSocket 数据帧结构

0                   1                   2                   3
 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-------+-+-------------+-------------------------------+
|F|R|R|R| opcode|M| Payload len |    Extended payload length    |
|I|S|S|S|  (4)  |A|     (7)     |             (16/64)           |
|N|V|V|V|       |S|             |   (if payload len==126/127)   |
| |1|2|3|       |K|             |                               |
+-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - +
|     Extended payload length continued, if payload len == 127  |
+ - - - - - - - - - - - - - - - +-------------------------------+
|                               |Masking-key, if MASK set to 1  |
+-------------------------------+-------------------------------+
| Masking-key (continued)       |          Payload Data         |
+-------------------------------- - - - - - - - - - - - - - - - +
:                     Payload Data continued ...                :
+ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +
|                     Payload Data continued ...                |
+---------------------------------------------------------------+

丢包处理机制

1. TCP 层保障机制（基础）

序列号与确认机制：每个数据包有唯一序列号，接收方需确认
重传计时器：发送方未收到ACK时重传数据
滑动窗口：动态调整发送速率
拥塞控制：慢启动、拥塞避免等算法

2. WebSocket 应用层保障（增强）

心跳机制（Ping/Pong）

// 服务端心跳实现
function setupHeartbeat(ws) {
  const heartbeatInterval = setInterval(() => {
    if (ws.readyState === ws.OPEN) {
      ws.ping();
      heartbeatTimeout = setTimeout(() => {
        ws.terminate(); // 超时断开
      }, 10000);
    }
  }, 30000);

  ws.on('pong', () => {
    clearTimeout(heartbeatTimeout);
  });

  ws.on('close', () => {
    clearInterval(heartbeatInterval);
  });
}

消息确认与重传

class ReliableWebSocket {
  constructor(url) {
    this.ws = new WebSocket(url);
    this.messageQueue = new Map();
    this.nextSeq = 1;
    
    this.ws.onmessage = (event) => {
      const msg = JSON.parse(event.data);
      if (msg.type === 'ACK') {
        this.messageQueue.delete(msg.seq);
      }
    };
  }

  sendReliable(data) {
    const seq = this.nextSeq++;
    const message = {
      seq,
      data,
      timestamp: Date.now(),
      retries: 0
    };
    
    this.messageQueue.set(seq, message);
    this.sendMessage(message);
    
    return new Promise((resolve) => {
      message.resolve = resolve;
    });
  }

  sendMessage(message) {
    this.ws.send(JSON.stringify(message));
    message.retryTimer = setTimeout(() => {
      if (this.messageQueue.has(message.seq)) {
        message.retries++;
        this.sendMessage(message);
      }
    }, 2000);
  }

  ack(seq) {
    this.ws.send(JSON.stringify({ type: 'ACK', seq }));
  }
}

3. 连接恢复机制

class ResilientWebSocket {
  constructor(url) {
    this.url = url;
    this.reconnectAttempts = 0;
    this.maxReconnectAttempts = 5;
    this.lastReceivedSeq = 0;
    this.connect();
  }

  connect() {
    this.ws = new WebSocket(this.url);
    
    this.ws.onopen = () => {
      this.reconnectAttempts = 0;
      this.resumeSession();
    };
    
    this.ws.onmessage = (event) => {
      const msg = JSON.parse(event.data);
      this.lastReceivedSeq = Math.max(this.lastReceivedSeq, msg.seq);
      // 处理消息...
    };
    
    this.ws.onclose = (event) => {
      if (event.code !== 1000 && this.reconnectAttempts < this.maxReconnectAttempts) {
        this.reconnect();
      }
    };
  }

  resumeSession() {
    this.ws.send(JSON.stringify({
      type: 'RESUME',
      lastSeq: this.lastReceivedSeq
    }));
  }

  reconnect() {
    const delay = Math.min(10000, 1000 * Math.pow(2, this.reconnectAttempts));
    this.reconnectAttempts++;
    
    setTimeout(() => this.connect(), delay);
  }
}