摘要：本文包括如下內(nèi)容協(xié)議第四章連接握手協(xié)議第五章數(shù)據(jù)幀庫源碼分析連接握手過程庫源碼分析數(shù)據(jù)幀解析過程參考協(xié)議深入探究本文對的概念定義解釋和用途等基礎(chǔ)知識不會涉及稍微偏干一點(diǎn)篇幅較長大約行閱讀需要耐心連接握手過程關(guān)于有一句很常見的話復(fù)用

本文包括如下內(nèi)容:

WebSocket協(xié)議第四章 - 連接握手

WebSocket協(xié)議第五章 - 數(shù)據(jù)幀

nodejs ws庫源碼分析 - 連接握手過程

nodejs ws庫源碼分析 - 數(shù)據(jù)幀解析過程

參考

WebSocket 協(xié)議深入探究

ws - github

本文對WebSocket的概念、定義、解釋和用途等基礎(chǔ)知識不會涉及, 稍微偏干一點(diǎn), 篇幅較長, markdown大約800行, 閱讀需要耐心

關(guān)于WebSocket有一句很常見的話: Websocket復(fù)用了HTTP的握手通道, 它具體指的是:

客戶端通過HTTP請求與WebSocket服務(wù)器協(xié)商升級協(xié)議, 協(xié)議升級完成后, 后續(xù)的數(shù)據(jù)交換則遵照WebSocket協(xié)議

首先由客戶端換發(fā)起協(xié)議升級請求, 根據(jù)WebSocket協(xié)議規(guī)范, 請求頭必須包含如下的內(nèi)容

GET / HTTP/1.1
Host: localhost:8080
Origin: http://127.0.0.1:3000
Connection: Upgrade
Upgrade: websocket
Sec-WebSocket-Version: 13
Sec-WebSocket-Key: w4v7O6xFTi36lq3RNcgctw==

請求行: 請求方法必須是GET, HTTP版本至少是1.1

請求必須含有Host

如果請求來自瀏覽器客戶端, 必須包含Origin

請求必須含有Connection, 其值必須含有"Upgrade"記號

請求必須含有Upgrade, 其值必須含有"websocket"關(guān)鍵字

請求必須含有Sec-Websocket-Version, 其值必須是13

請求必須含有Sec-Websocket-Key, 用于提供基本的防護(hù), 比如無意的連接

服務(wù)器返回的響應(yīng)頭必須包含如下的內(nèi)容

HTTP/1.1 101 Switching Protocols
Connection:Upgrade
Upgrade: websocket
Sec-WebSocket-Accept: Oy4NRAQ13jhfONC7bP8dTKb4PTU=

響應(yīng)行: HTTP/1.1 101 Switching Protocols

響應(yīng)必須含有Upgrade, 其值為"weboscket"

響應(yīng)必須含有Connection, 其值為"Upgrade"

響應(yīng)必須含有Sec-Websocket-Accept, 根據(jù)請求首部的Sec-Websocket-key計(jì)算出來

規(guī)范提到:

Sec-WebSocket-Key值由一個隨機(jī)生成的16字節(jié)的隨機(jī)數(shù)通過base64（見RFC4648的第四章）編碼得到的

例如, 隨機(jī)選擇的16個字節(jié)為:

// 十六進(jìn)制 數(shù)字1~16
0x01 0x02 0x03 0x04 0x05 0x06 0x07 0x08 0x09 0x0a 0x0b 0x0c 0x0d 0x0e 0x0f 0x10

通過base64編碼后值為: AQIDBAUGBwgJCgsMDQ4PEA==

測試代碼如下:

const list = Array.from({ length: 16 }, (v, index) => ++index)
const key = Buffer.from(list)
console.log(key.toString("base64"))
// AQIDBAUGBwgJCgsMDQ4PEA==

而Sec-WebSocket-Accept值的計(jì)算方式為:

將Sec-Websocket-Key的值和258EAFA5-E914-47DA-95CA-C5AB0DC85B11拼接

通過SHA1計(jì)算出摘要, 并轉(zhuǎn)成base64字符串

此處不需要糾結(jié)神奇字符串258EAFA5-E914-47DA-95CA-C5AB0DC85B11, 它就是一個GUID, 沒準(zhǔn)兒是寫RFC的時(shí)候隨機(jī)生成的

測試代碼如下:

const crypto = require("crypto")

function hashWebSocketKey (key) {
  const GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"

  return crypto.createHash("sha1")
    .update(key + GUID)
    .digest("base64")
}

console.log(hashWebSocketKey("w4v7O6xFTi36lq3RNcgctw=="))
// Oy4NRAQ13jhfONC7bP8dTKb4PTU=

前面簡單提到他的作用為: 提供基礎(chǔ)的防護(hù), 減少惡意連接, 進(jìn)一步闡述如下:

Key可以避免服務(wù)器收到非法的WebSocket連接, 比如http請求連接到websocket, 此時(shí)服務(wù)端可以直接拒絕

Key可以用來初步確保服務(wù)器認(rèn)識ws協(xié)議, 但也不能排除有的http服務(wù)器只處理Sec-WebSocket-Key, 并不實(shí)現(xiàn)ws協(xié)議

Key可以避免反向代理緩存

在瀏覽器中發(fā)起ajax請求, Sec-Websocket-Key以及相關(guān)header是被禁止的, 這樣可以避免客戶端發(fā)送ajax請求時(shí), 意外請求協(xié)議升級

最終需要強(qiáng)調(diào)的是: Sec-WebSocket-Key/Accept并不是用來保證數(shù)據(jù)的安全性, 因?yàn)槠溆?jì)算/轉(zhuǎn)換公式都是公開的, 而且非常簡單, 最主要的作用是預(yù)防一些意外的情況

WebSocket通信的最小單位是幀, 由一個或多個幀組成一條完整的消息, 交換數(shù)據(jù)的過程中, 發(fā)送端和接收端需要做的事情如下:

發(fā)送端: 將消息切割成多個幀, 并發(fā)送給服務(wù)端

接收端: 接受消息幀, 并將關(guān)聯(lián)的幀重新組裝成完整的消息

數(shù)據(jù)幀格式作為核心內(nèi)容, 一眼看去似乎難以理解, 但本文作者下死命令了, 必須理解, 沖沖沖

FIN: 占1bit

0表示不是消息的最后一個分片

1表示是消息的最后一個分片

RSV1, RSV2, RSV3: 各占1bit, 一般情況下全為0, 與Websocket拓展有關(guān), 如果出現(xiàn)非零的值且沒有采用WebSocket拓展, 連接出錯

Opcode: 占4bit

%x0: 表示本次數(shù)據(jù)傳輸采用了數(shù)據(jù)分片, 當(dāng)前數(shù)據(jù)幀為其中一個數(shù)據(jù)分片

%x1: 表示這是一個文本幀

%x2: 表示這是一個二進(jìn)制幀

%x3-7: 保留的操作代碼, 用于后續(xù)定義的非控制幀

%x8: 表示連接斷開

%x9: 表示這是一個心跳請求(ping)

%xA: 表示這是一個心跳響應(yīng)(pong)

%xB-F: 保留的操作代碼, 用于后續(xù)定義的非控制幀

Mask: 占1bit

0表示不對數(shù)據(jù)載荷進(jìn)行掩碼異或操作

1表示對數(shù)據(jù)載荷進(jìn)行掩碼異或操作

Payload length: 占7或7+16或7+64bit

0~125: 數(shù)據(jù)長度等于該值

126: 后續(xù)的2個字節(jié)代表一個16位的無符號整數(shù), 值為數(shù)據(jù)的長度

127: 后續(xù)的8個字節(jié)代表一個64位的無符號整數(shù), 值為數(shù)據(jù)的長度

Masking-key: 占0或4bytes

1: 攜帶了4字節(jié)的Masking-key

0: 沒有Masking-key

掩碼的作用并不是防止數(shù)據(jù)泄密,而是為了防止早期版本協(xié)議中存在的代理緩存污染攻擊等問題

payload data: 載荷數(shù)據(jù)

我想如果知道byte和bit的區(qū)別, 這部分就沒問題- -

WebSocket的每條消息可能被切分成多個數(shù)據(jù)幀, 當(dāng)接收到一個數(shù)據(jù)幀時(shí),會根據(jù)FIN值來判斷, 是否為最后一個數(shù)據(jù)幀

數(shù)據(jù)幀傳遞示例:

FIN=0, Opcode=0x1: 發(fā)送文本類型, 消息還沒有發(fā)送完成,還有后續(xù)幀

FIN=0, Opcode=0x0: 消息沒有發(fā)送完成, 還有后續(xù)幀, 接在上一條后面

FIN=1, Opcode=0x0: 消息發(fā)送完成, 沒有后續(xù)幀, 接在上一條后面組成完整消息

雖然之前用的都是socket.io, 偶然發(fā)現(xiàn)了ws, 使用量竟然還挺大, 周下載量是socket.io的六倍

在NodeJS中, 每當(dāng)遇到協(xié)商升級請求時(shí), 就會觸發(fā)http模塊的upgrade事件, 這便是實(shí)現(xiàn)WebSocketServer的切入點(diǎn), 原生示例代碼如下:

// 創(chuàng)建 HTTP 服務(wù)器。
const srv = http.createServer( (req, res) => {
  res.writeHead(200, { "Content-Type": "text/plain" });
  res.end("響應(yīng)內(nèi)容");
});
srv.on("upgrade", (req, socket, head) => {
  // 特定的處理, 以實(shí)現(xiàn)Websocket服務(wù)
});

并且, 在一般的使用中, 都是在一個已有的httpServer基礎(chǔ)上進(jìn)行拓展, 以實(shí)現(xiàn)WebSocket, 而不是創(chuàng)建一個獨(dú)立的WebSocketServer

在一個已有httpServer的基礎(chǔ)上, ws使用的實(shí)例代碼為

const http = require("http");
const WebSocket = require("ws");

const server = http.createServer();
const wss = new WebSocket.Server({ server });

server.listen(8080);

已有的httpServer作為參數(shù)傳給了WebSocket.Server構(gòu)造函數(shù), 所以源碼分析的核心切入點(diǎn)為:

new WebSocket.Server({ server });

通過這個切入點(diǎn), 就可以完整復(fù)現(xiàn)連接握手的過程

因?yàn)?b>httpServer已作為參數(shù)傳遞進(jìn)來, 因此其構(gòu)造函數(shù)變得十分簡單:

class WebSocketServer extends EventEmitter {
  constructor(options, callback) {
    super()
    // 在提供了http server的基礎(chǔ)上, 代碼可以簡化為
    if (options.server) {
      this._server = options.server
    }
    // 監(jiān)聽事件
    if (this._server) {
      this._removeListeners = addListeners(this._server, {
        listening: this.emit.bind(this, "listening"),
        error: this.emit.bind(this, "error"),
        // 核心
        upgrade: (req, socket, head) => {
          // 下一步切入點(diǎn)
          this.handleUpgrade(req, socket, head, (ws) => {
            this.emit("connection", ws, req)
          })
        }
      })
    }
  }
}

// 這是一段非常帶秀的代碼, 在綁定多個事件監(jiān)聽器的同時(shí)返回一個移除多個事件監(jiān)聽器的函數(shù)
function addListeners(server, map) {
  for (const event of Object.keys(map)) server.on(event, map[event]);

  return function removeListeners() {
    for (const event of Object.keys(map)) {
      server.removeListener(event, map[event]);
    }
  };
}

可以看到, 在構(gòu)造函數(shù)中, 為httpServer注冊了upgrade事件的監(jiān)聽器, 觸發(fā)時(shí), 會執(zhí)行this.handleUpgrade函數(shù), 這便是下一步的方向

這個函數(shù)主要用來過濾掉不合法的請求, 檢查的內(nèi)容包括:

Sec-WebSocket-Key值

Sec-WebSocket-Version值

WebSocket請求的路徑

關(guān)鍵代碼如下:

const keyRegex = /^[+/0-9A-Za-z]{22}==$/;

handleUpgrade(req, socket, head, cb) {
  socket.on("error", socketOnError)

  // 獲取sec-websocket-key
  const key = req.headers["sec-websocket-key"] !== undefined
    ? req.headers["sec-websocket-key"]
    : false

  // 獲取sec-websocket-version
  const version = +req.headers["sec-websocket-version"]

  // 獲取協(xié)議拓展, 本篇不涉及
  const extensions = {};

  // 對于不合法的請求, 中斷握手
  if (
    req.method !== "GET" ||
    req.headers.upgrade.toLowerCase() !== "websocket" ||
    !key ||
    !keyRegex.test(key) ||
    (version !== 8 && version !== 13) ||
    // 該函數(shù)是對Websocket請求路徑的判斷, 與option.path相關(guān), 不展開
    !this.shouldHandle(req)
  ) {
    return abortHandshake(socket, 400)
  }

  // 對于合法的請求, 給它升級!
  this.completeUpgrade(key, extensions, req, socket, head, cb)
}

對于不合法的請求, 直接400 bad request了, abortHandshake如下:

const {  STATUS_CODES } = require("http");

function abortHandshake(socket, code, message, headers) {
  // net.Socket 也是雙工流，因此它既可讀也可寫
  if (socket.writable) {
    message = message || STATUS_CODES[code];
    headers = {
      Connection: "close",
      "Content-type": "text/html",
      "Content-Length": Buffer.byteLength(message),
      ...headers
    };

    socket.write(
      `HTTP/1.1 ${code} ${STATUS_CODES[code]}
` +
        Object.keys(headers)
          .map((h) => `${h}: ${headers[h]}`)
          .join("
") +
        "

" +
        message
    );
  }
  // 移除handleUpgrade中添加的error監(jiān)聽器
  socket.removeListener("error", socketOnError);
  // 確保在該 socket 上不再有 I/O 活動
  socket.destroy();
}

如果一切順利, 我們來到completeUpgrade函數(shù)

這個函數(shù)主要用來, 返回正確的響應(yīng), 觸發(fā)相關(guān)的事件, 記錄值等, 代碼比較簡單

const { createHash } = require("crypto");
const { GUID } = require("./constants");
const WebSocket = require("./websocket");

function completeUpgrade(key, extensions, req, socket, head, cb) {
  // Destroy the socket if the client has already sent a FIN packet.
  if (!socket.readable || !socket.writable) return socket.destroy()

  // 生成sec-websocket-accept
  const digest = createHash("sha1")
    .update(key + GUID)
    .digest("base64");

  // 組裝Headers
  const headers = [
    "HTTP/1.1 101 Switching Protocols",
    "Upgrade: websocket",
    "Connection: Upgrade",
    `Sec-WebSocket-Accept: ${digest}`
  ];
  // 創(chuàng)建一個Websocket實(shí)例
  const ws = new Websocket(null)

  this.emit("headers", headers, req);
  // 返回響應(yīng)
  socket.write(headers.concat("
").join("
"));
  socket.removeListener("error", socketOnError);

  // 下一步切入點(diǎn)
  ws.setSocket(socket, head, this.options.maxPayload);

  // 通過Set記錄處于連接狀態(tài)的客戶端
  if (this.clients) {
    this.clients.add(ws);
    ws.on("close", () => this.clients.delete(ws));
  }
  // 觸發(fā)connection事件
  cb(ws);
}

到這里, 就完成了整個握手階段, 但還沒涉及到對數(shù)據(jù)幀的處理

上一章末尾, 啟示下文的代碼為completeUpgrade中的:

ws.setSocket(socket, head, this.options.maxPayload);

進(jìn)入WebSocket類中的setSocket方法, 關(guān)于數(shù)據(jù)幀處理代碼主要可以簡化為:

Class WebSocket extends EventEmitter {
  ...
  setSocket(socket, head, maxPayload) {
    // 實(shí)例化一個可寫流, 用于處理數(shù)據(jù)幀
    const receiver = new Receiver(
      this._binaryType,
      this._extensions,
      maxPayload
    );
    receiver[kWebSocket] = this;
    socket.on("data", socketOnData);
  }
}
function socketOnData(chunk) {
  if (!this[kWebSocket]._receiver.write(chunk)) {
    this.pause();
  }
}

此處忽略了很多事件處理, 例如error, end, close等, 因?yàn)樗麄兣c本文目標(biāo)無關(guān), 對于一些API, 也不做介紹

所以核心切入點(diǎn)為Receiver類, 它就是用于處理數(shù)據(jù)幀的核心

Receiver類繼承自可寫流, 還需要明確兩點(diǎn)基本概念:

stream所有的流都是EventEmitter的實(shí)例

實(shí)現(xiàn)可寫流需要實(shí)現(xiàn)writable._write方法, 該方法供內(nèi)部使用

const { Writable } = require("stream")

class Recevier extends Writable {
  constructor(binaryType, extensions, maxPayload) {
    super()

    this._binaryType = binaryType || BINARY_TYPES[0]; // nodebuffer
    this[kWebSocket] = undefined; // WebSocket實(shí)例的引用
    this._extensions = extensions || {}; // WebSocket協(xié)議拓展
    this._maxPayload = maxPayload | 0; // 100 * 1024 * 1024

    this._bufferedBytes = 0; // 記錄buffer長度
    this._buffers = []; // 記錄buffer數(shù)據(jù)

    this._compressed = false; // 是否壓縮
    this._payloadLength = 0; // 數(shù)據(jù)幀 PayloadLength
    this._mask = undefined; // 數(shù)據(jù)幀Mask Key
    this._fragmented = 0; // 數(shù)據(jù)幀是否分片
    this._masked = false; // 數(shù)據(jù)幀 Mask
    this._fin = false; // 數(shù)據(jù)幀 FIN
    this._opcode = 0;  // 數(shù)據(jù)幀 Opcode

    this._totalPayloadLength = 0; // 載荷總長度
    this._messageLength = 0; // 載荷總長度, 與this._compressed有關(guān)
    this._fragments = []; // 載荷分片記錄數(shù)組

    this._state = GET_INFO; // 標(biāo)志位, 用于startLoop函數(shù)
    this._loop = false; // 標(biāo)志位, 用于startLoop函數(shù)
  }

  _write(chunk, encoding, cb) {
    if (this._opcode === 0x08 && this._state == GET_INFO) return cb();

    this._bufferedBytes += chunk.length;
    this._buffers.push(chunk);
    this.startLoop(cb);
  }
}

可以看到, 每當(dāng)收到新的數(shù)據(jù)幀, 就會將其記錄在_buffers數(shù)組中, 并立即開始解析流程startLoop

startLoop(cb) {
  let err;
  this._loop = true;

  do {
    switch (this._state) {
      case GET_INFO:
        err = this.getInfo();
        break;
      case GET_PAYLOAD_LENGTH_16:
        err = this.getPayloadLength16();
        break;
      case GET_PAYLOAD_LENGTH_64:
        err = this.getPayloadLength64();
        break;
      case GET_MASK:
        this.getMask();
        break;
      case GET_DATA:
        err = this.getData(cb);
        break;
      default:
        // `INFLATING`
        this._loop = false;
        return;
    }
  } while (this._loop);

  cb(err);
}

解析流程很簡單:

getInfo首先解析FIN, RSV, OPCODE, MASK, PAYLOAD LENGTH等數(shù)據(jù)

因?yàn)?b>payload length分為三種情況(具體后面敘述, 此處只列出分支):

0~125: 調(diào)用haveLength方法

126: 先觸發(fā)getPayloadLength16方法, 再調(diào)用haveLength方法

127: 先出法getPayloadLength64方法, 再調(diào)用haveLength方法

haveLength方法中, 如果存在掩碼(mask), 先調(diào)用getMask方法, 再調(diào)用getData方法

整體流程和狀態(tài)通過this._loop和this._state控制, 比較直觀

按理說第一步應(yīng)該分析getInfo方法, 不過里面涉及到了consume方法, 這個函數(shù)提供了一種簡潔的方式消費(fèi)已獲取的Buffer, 這個函數(shù)接受一個參數(shù)n, 代表需要消費(fèi)的字節(jié)數(shù), 最后返回消費(fèi)的字節(jié)

假如需要獲得數(shù)據(jù)幀的第一個字節(jié)的數(shù)據(jù)(包含了 FIN + RSV + OPCODE), 只需要通過this.consume(1)即可

記錄值this._buffers是一個buffer數(shù)組, 最開始, 里面存放完整的數(shù)據(jù)幀, 隨著消費(fèi)的進(jìn)行, 數(shù)據(jù)則會逐漸變小, 那么每次消費(fèi)存在三種可能:

消費(fèi)的字節(jié)數(shù)恰好等于一個chunk的字節(jié)數(shù)

消費(fèi)的字節(jié)數(shù)小于一個chunk的字節(jié)數(shù)

消費(fèi)的字節(jié)數(shù)大于一個chunk的字節(jié)數(shù)

對于第一種情況, 只需要移出 + 返回即可

if (n === this._buffers[0].length) return this._buffers.shift()

對于第二種情況, 只需要裁剪 + 返回即可

if (n < this._buffers[0].length) {
  const buf = this._buffers[0]
  this._buffers[0] = buf.slice(n)
  return buf.slice(0, n)
}

對于第三種情況, 會稍微復(fù)雜一點(diǎn), 首先我們要申請一個大小為需要消費(fèi)字節(jié)數(shù)的buffer空間, 用于存儲返回的buffer

// buffer空間是否初始化并不重要, 因?yàn)樽罱K他都會被全部覆蓋
const dst = Buffer.allocUnsafe(n)

在這種情況中, 可以保證他的長度大于第一個chunk, 但不能確定在消費(fèi)一個chunk之后, 是否還大于第一個chunk(消費(fèi)之后索引前移), 因此需要循環(huán)

// do...while可以避免一次無意義判斷, 首先執(zhí)行一次循環(huán)體, 再判斷條件
do {
  const buf = this._buffers[0]

  // 如果長度大于第一個chunk, 移除 + 復(fù)制即可
  if (n >= buf.length) {
    this._buffers.shift().copy(dst, dst.length - n);
  }
  // 如果長度小于一個chunk, 裁剪 + 復(fù)制即可
  else {
    // buf.copy這個api就自己復(fù)習(xí)一下嗷
    buf.copy(dst, dst.length - n, 0, n);
    this._buffers[0] = buf.slice(n);
  }
  n -= buf.length;
} while (n > 0)

一個最小的數(shù)據(jù)幀必須包含如下的數(shù)據(jù):

FIN (1 bit) + RSV (3 bit) + OPCODE (4 bit) + MASK (1 bit) + PAYLOADLENGTH (7 bit)

最少2個字節(jié), 因此少于兩個字節(jié)的數(shù)據(jù)幀是錯誤的, 簡化的getInfo如下

getInfo() {
  if (this._bufferedBytes < 2) {
    this._loop = false
    return
  }
  const buf = this.consume(2)

  // 只保留了數(shù)據(jù)幀中的幾個關(guān)鍵數(shù)據(jù)
  this._fin = (buf[0] & 0x80) === 0x80
  this._opcode = buf[0] & 0x0f
  this._payloadLength = buf[1] & 0x7f
  this._masked = (buf[1] & 0x80) === 0x80

  // 對應(yīng)Payload Length的三種情況
  if (this._payloadLength === 126) this._state = GET_PAYLOAD_LENGTH_16
  else if (this._payloadLength === 127) this._state = GET_PAYLOAD_LENGTH_64
  else return this.haveLength()
}

此處的核心就是按位于運(yùn)算符&的含義, 先以FIN為例, FIN在數(shù)據(jù)幀中處于第一個bit

// FIN的值用[]指代, X代表第一個字節(jié)中的后續(xù)bit
[]xxxxxxx
// 十六進(jìn)制數(shù)0x80代表二進(jìn)制
10000000
// 兩者按位與, 結(jié)果與后面7個bit無關(guān)
[]0000000
// 因此, 只需要比較[]0000000 和 10000000是否相等即可, 簡化即得到
this._fin = (buf[0] & 0x80) === 0x80

OPCODE和PAYLOAD LENGTH同理

// OPCODE處于第一個字節(jié)的后四位, 與0000 1111按位與即可
xxxx[][][][] & 0000 1111 (也就是0x0f)

// PAYLOAD LENGTH處于第二個字節(jié)的后七為, 與0111 1111按位于即可
x[][][][][][][][] & 0111 1111 (也就是0x7f)

三種情況如下:

0-125: 載荷實(shí)際長度就是0-125之間的某個數(shù)

126: 載荷實(shí)際長度為隨后2個字節(jié)代表的一個16位的無符號整數(shù)的數(shù)值

127: 載荷實(shí)際長度為隨后8個字節(jié)代表的一個64位的無符號整數(shù)的數(shù)值

可能聽起來比較繞, 看代碼, 以126分支為例:

getPayloadLength16() {
  if (this._bufferedBytes < 2) {
    this._loop = false;
    return;
  }

  this._payloadLength = this.consume(2).readUInt16BE(0);
  return this.haveLength();
}

可以看到, 處理長度的核心為readUInt16BE(0), 這便涉及到大小端了:

大端(Big endian)認(rèn)為第一個字節(jié)是最高位字節(jié), 和我們對十進(jìn)制數(shù)字大小的認(rèn)知相似

小端(Little endian)認(rèn)為第一個字節(jié)是最低位字節(jié)

那么, 規(guī)范中提到的隨后2個字節(jié)代表的一個16位的無符號整數(shù)的數(shù)值, 自然指的是大端了

大端 vs 小端對比:

// 假設(shè)后面兩個字節(jié)二進(jìn)制值為
1111 1111 0000 0001
// 轉(zhuǎn)為十六進(jìn)制為
0xff 0x01
// 大端輸出 65281
console.log(Buffer.from([0xff, 0x01]).readUInt16BE(0).toString(10))
// 小端輸出 511
console.log(Buffer.from([0xff, 0x01]).readUInt16LE(0).toString(10))

除此之外, 7 + 64的模式還有一點(diǎn)額外的處理, 代碼如下:

getPayloadLength64() {
  if (this._bufferedBytes < 8) {
    this._loop = false;
    return;
  }

  const buf = this.consume(8);
  const num = buf.readUInt32BE(0);

  //
  // The maximum safe integer in JavaScript is 2^53 - 1. An error is returned
  // if payload length is greater than this number.
  //
  if (num > Math.pow(2, 53 - 32) - 1) {
    this._loop = false;
    return error(
      RangeError,
      "Unsupported WebSocket frame: payload length > 2^53 - 1",
      false,
      1009
    );
  }

  this._payloadLength = num * Math.pow(2, 32) + buf.readUInt32BE(4);
  return this.haveLength();
}

在獲得載荷之前, 如果getInfo中mask為1, 需要進(jìn)行getMask操作, 獲取Mask Key(一共四個字節(jié))

getMask() {
  if (this._bufferedBytes < 4) {
    this._loop = false;
    return;
  }

  this._mask = this.consume(4);
  this._state = GET_DATA;
}

getData源碼簡化為如下

getData(cb) {
  // data為 Buffer.alloc(0)
  let data = EMPTY_BUFFER;

  // 消費(fèi)payload
  data = this.consume(this._payloadLength)
  // 如果有mask, 根據(jù)mask key進(jìn)行解碼, 此處不展開
  if (this._masked) unmask(data, this._mask)
  // 將其記錄進(jìn)分片數(shù)組
  this._fragments.push(data)
  // 如果該數(shù)據(jù)幀表示: 連接斷開, 心跳請求, 心跳響應(yīng)
  if (this._opcode > 0x07) return this.controlMessage(data)
  // 如果該數(shù)據(jù)幀表示: 數(shù)據(jù)分片、文本幀、二進(jìn)制幀
  return this.dataMessage()
}

接著分析dataMessage()函數(shù), 它用于將多個幀的數(shù)據(jù)合并, 簡化之后也比較簡單

dataMessage() {
  if (this._fin) {
    const messageLength = this._messageLength
    const fragments = this._fragments

    const buf = concat(fragments, messageLength)
    this.emit("message", buf.toString())
  }
}
// 簡明易懂哦, 不解釋啦
function concat(list, totalLength) {
  if (list.length === 0) return EMPTY_BUFFER;
  if (list.length === 1) return list[0];

  const target = Buffer.allocUnsafe(totalLength);
  let offset = 0;

  for (let i = 0; i < list.length; i++) {
    const buf = list[i];
    buf.copy(target, offset);
    offset += buf.length;
  }

  return target;
}

本文篇幅較長且并不是面試題那種小塊的知識點(diǎn), 閱讀急需耐心, 已盡量避免貼大段代碼, 能看到這里我都想給你打錢了

通過本篇分析, 完整的介紹以及復(fù)現(xiàn)了WebSocket中的兩個關(guān)鍵階段:

連接握手階段

數(shù)據(jù)交換極端

個人認(rèn)為最關(guān)鍵便是: 涉及到了對Node.js的buffer模塊以及stream模塊的使用, 這也是收獲最大的一部分