前言

為什么寫這篇文章？

對于應用協議的了解，相信大部分👨‍🎓始終停留在使用上，可能讀了如《圖解HTTP》、《計算機網絡》一類的書籍，了解了更深層次的理論，但理論始終是理論，我們很難有機會能在工作場景里面去觸碰到協議的實現上，對應用層協議的理解是片面的。

WebSocket 的實現非常適合前端的同學學習，通過了解 WebSocket Node版實現，站在更高的維度上去看待這些應用層協議，無論是WebRTC，還是HTTP都好，有了對一種協議實現的整體思路，就有了面對各種協議的技術自信。

WebSocket簡介

WebSocket 是 HTML5 開始提供的一種在單個 TCP 連接上進行全雙工(full-duplex)通訊的協議。沒有了 Request 和 Response 的概念，兩者地位完全平等，連接一旦建立，就建立了真•持久性連接，雙方可以隨時向對方發送數據。

注：全雙工（Full Duplex）是一種通信方式，指通信的雙方可以同時發送和接收數據，而且在同一時刻，發送和 接收是獨立進行的。

概念老生常談了，不妨思考兩個問題：

WebSocket 為什么能進行全雙工通信？HTTP 卻不行？

HTTP是非持久化連接，每次客戶端向服務器發送請求時，都需要建立一個新的 TCP 連接。這個連接在響應結束后就會被關閉，不保留在系統中，而 websocket 會保留，所以可以繼續保持通信。

HTTP 為什么是非持久化連接而 WebSocket 是持久化連接？

HTTP 協議的設計初衷是為了傳輸靜態文本信息，如 HTML、CSS、JS 等等，傳輸靜態文本大部分時間連接并不會被頻繁地打開和關閉，所以使用持久化連接帶來的復雜度可能會超過它的收益。而不像 websocket 更多是為了實時通信的場景，所以采用持久化連接。

WebSocket握手過程

我們經常聽到一種說法，WebSocket 基于 HTTP，實際上只有在建立握手時，數據是通過 HTTP 傳輸的。但是建立之后，在真正傳輸時候是不需要 HTTP協議。握手具體過程如圖所示：

注：websocket 會保留 HTTP 握手后的 socket 連接，后續即可利用這個 socket 進行通訊，無需再關注 HTTP。

websocket 為什么采用 HTTP 握手？

WebSocket 是相對較新的協議，可能并不是所有的網絡設備和服務器都支持。因此，在瀏覽器請求服務器進行 WebSocket 握手時，使用基于 HTTP 協議的握手方式可以避免協議兼容性問題。

WebSocket實現原理

WebSocket數據幀說明

WebSocket 以幀的形式進行數據傳輸，幀組成包括以下幾個部分：

頭部（Header）：包括了一些控制信息，如 FIN、RSV1、RSV2、RSV3、Opcode、MASK、Payload Length 以及 Masking Key 等字段，用于描述該幀的類型、長度以及是否經過掩碼操作等信息。它有 2~14 個字節不等，其中前 2 個字節是必需的。
掩碼（Masking Key）：用于對載荷（Payload）進行加密解密操作，它的長度固定為 4 個字節，如果 MASK 標志被設置為 1，則該字段必須存在。
載荷（Payload）：包含了應用層發送的數據，具體內容由 Opcode 字段指定的數據類型決定。如果有 MASK 標志，則需要對其進行解碼。

各字段含義如表格所示：

字段	含義	長度
FIN	是否為最后一幀	1 bit
RSV	預留位，方便后續拓展協議	3 bit
opcode	解釋 payload data 的用途	4 bit
MASK	定義“payload data”是否被添加掩碼	1 bit
payload data length	數據長度	7 bit 7+16 bit 7+64 bit
Serial Number	序列號	16bit
Masking-key	掩碼	32bit
payload data	傳輸數據	payload data length

把數據幀組成搞清楚，可以說 websocket 你就了解了一大半，協議實現里大部分操作都是對于數據幀的處理。

構造幀

websocket 的數據是以幀的形式傳輸，那么我們就需要了解如何構造幀。構造幀只是聽起來很復雜，構造一個數據幀我們只需要遵守協議規則填寫即可，按照WebSocket的協議標準，構造一個最短數據幀我們只需要三個字節就能完成，構成如表格所示下：

字節編號	填入內容
1	FIN、RSV、opcode
2	MASK、payload data length
3	payload data

代碼實現：

注：本質就是做一些字節拼接操作，把對應的標識放到對應的位置即可。

數據傳輸

了解往構造幀的過程，那 websocket 是如何把幀發送出去的？

WebSocket 在握手過程中會保留 HTTP 握手后的 socket 連接，這在前面有提到，所以我們可以通過這個 socket 連接進行數據的傳輸。

代碼實現如下：

心跳機制

在連接過中，防止連接因長時間無數據傳輸而被提前關閉，WebSocket 還引入了心跳機制，原理可以概括為：定期發送心跳包，以確認客戶端與服務器的連接狀態，并避免連接因長時間空閑而被中斷。代碼實現如下：

mini-ws

以下為一個 websocket server 的簡易實現（代碼來源）：

測試代碼：

var crypto = require("crypto");
var { EventEmitter } = require("events");
var MAX_FRAME_SIZE = 1024; // 最長長度限制
var MAGIC_STRING = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";

/**
 *  數據類型操作碼 TEXT 字符串
 *  BINARY 二進制數據 常用來保存照片
 *  PING,PONG 用作心跳檢測
 *  CLOSE 關閉連接的數據幀 (有很多關閉連接的代碼 1001，1009,1007,1002)
 */
var OPCODES = {
    CONTINUE: 0,
    TEXT: 1,
    BINARY: 2,
    CLOSE: 8,
    PING: 9,
    PONG: 10,
};

var hashWebSocketKey = function (key) {
    var sha1 = crypto.createHash("sha1");
    sha1.update(key + MAGIC_STRING, "ascii");
    return sha1.digest("base64");
};
/**
 * 解掩碼
 * @param maskBytes 掩碼數據
 * @param data payload
 * @returns {Buffer}
 */
var unmask = function (maskBytes, data) {
    var payload = Buffer.alloc(data.length);
    for (var i = 0; i < data.length; i++) {
        payload[i] = maskBytes[i % 4] ^ data[i];
    }
    return payload;
};

/**
 * 編碼數據
 * @param opcode 操作碼
 * @param payload   數據
 * @returns {*}
 */
var encodeMessage = function (opcode, payload, isFinal = true) {
    var buf;
    var b1 = (isFinal ? 0x80 : 0x00) | opcode;
    var b2;
    var length = payload.length;
    if (length < 126) {
        buf = Buffer.alloc(payload.length + 2 + 0);
        b2 |= length;
        //buffer ,offset
        buf.writeUInt8(b1, 0); //讀前8bit
        buf.writeUInt8(b2, 1); //讀8―15bit
        payload.copy(buf, 2); //復制數據,從2(第三)字節開始
    } else if (length < 1 << 16) {
        buf = Buffer.alloc(payload.length + 2 + 2);
        b2 |= 126;
        buf.writeUInt8(b1, 0);
        buf.writeUInt8(b2, 1);
        buf.writeUInt16BE(length, 2);
        payload.copy(buf, 4);
    } else {
        buf = Buffer.alloc(payload.length + 2 + 8);
        b2 |= 127;
        buf.writeUInt8(b1, 0);
        buf.writeUInt8(b2, 1);
        buf.writeUInt32BE(0, 2);
        buf.writeUInt32BE(length, 6);
        payload.copy(buf, 10);
    }

    return buf;
};

class WebSocket extends EventEmitter {
    constructor(req, socket, upgradeHead) {
        super();
        var resKey = hashWebSocketKey(req.headers["sec-websocket-key"]);

        // 構造響應頭
        var resHeaders = [
            "HTTP/1.1 101 Switching Protocols",
            "Upgrade: websocket",
            "Connection: Upgrade",
            "Sec-WebSocket-Accept: " + resKey,
        ]
            .concat("", "")
            .join("\r\n");

        socket.on("data", (data) => {
            this.buffer = Buffer.concat([this.buffer, data]);
            while (this._processBuffer()) {}
        });

        socket.on("close", (had_error) => {
            if (!this.closed) {
                this.emit("close", 1006);
                this.closed = true;
            }
        });

        socket.write(resHeaders);

        this.socket = socket;
        this.buffer = Buffer.alloc(0);
        this.closed = false;
        this.frames = Buffer.alloc(0);
        this.frameOpcode = 0;
        this.keepLiveTimer = null;
    }
    /*
 發送數據函數
 * */
    send(obj) {
        var opcode;
        var payload;
        // 如果是二進制
        if (Buffer.isBuffer(obj)) {
            opcode = OPCODES.BINARY;
            payload = obj;
        } else if (typeof obj) {
            // 承載的文本內容
            opcode = OPCODES.TEXT;
            //創造一個utf8的編碼，可以被編碼為字符串
            payload = Buffer.from(obj, "utf8");
        } else {
            throw new Error("cannot send object.Must be string of Buffer");
        }

        this._doSend(opcode, payload);
    }

    // 默認 45 秒 保持發送心跳
    keepLive(timeout = 45000) {
        var self = this;
        function keepit() {
            self._doSend(OPCODES.PING, Buffer.from("ping"));
            console.log("server send ping...");
            // 在關閉連接的情況下就不再需要發送 ping 請求了
            if (!self.closed) {
                self.keepLiveTimer = setTimeout(keepit, timeout);
            }
        }
        keepit();
    }

    /*
 關閉連接函數
 * */
    close(code, reason) {
        var opcode = OPCODES.CLOSE;
        var buffer;
        if (code) {
            buffer = Buffer.alloc(Buffer.byteLength(reason) + 2);
            buffer.writeUInt16BE(code, 0);
            buffer.write(reason, 2);
        } else {
            buffer = Buffer.alloc(0);
        }
        this._doSend(opcode, buffer);
        this.closed = true;
    }

    _processBuffer() {
        var buf = this.buffer;
        if (buf.length < 2) {
            return;
        }
        var idx = 2;
        var byte1 = buf.readUInt8(0); // 讀取數據幀的前 8 bit
        var FIN = byte1 & 0x80; // 如果為0x80，則標志傳輸結束，獲取高位 bit
        var opcode = byte1 & 0x0f; //截取第一個字節的后 4 位，即 opcode 碼

        // 如果是 0 的話，說明是延續幀，需要保存好 opCode
        if (!FIN) {
            this.frameOpcode = opcode || this.frameOpcode; // 確保不為 0;
        }

        var byte2 = buf.readUInt8(1); // 讀取數據幀第二個字節
        var MASK = byte2 & 0x80; // 判斷是否有掩碼，客戶端必須要有，獲取高位 bit
        var length = byte2 & 0x7f; //獲取length屬性，也是小于126數據長度的數據真實值
        if (length > 125) {
            if (buf.length < 8) {
                return; // 如果大于125，而字節數小于 8，則顯然不合規范要求
            }
        }
        if (length === 126) {
            //獲取的值為126 ，表示后兩個字節（16位）用于表示數據長度
            length = buf.readUInt16BE(2); // 讀取 16bit 的值
            idx += 2; // +2
        } else if (length === 127) {
            //獲取的值為 127 ，表示后 8 個字節（64位）用于表示數據長度，其中高 4 字節是 0
            var highBits = buf.readUInt32BE(2); //(1/0)1111111，切記 MSB 最高位是 0
            if (highBits != 0) {
                this.close(1009, ""); //1009 關閉代碼，說明數據太大； 協議里是支持 63 位長度，不過這里我們自己實現的話，只支持 32 位長度，防止數據過大；
            }
            length = buf.readUInt32BE(6); // 從第 6 到第 10 個字節（32位）為真實存放的數據長度
            idx += 8;
        }
        if (buf.length < idx + 4 + length) {
            //不夠長 4為掩碼字節數
            return;
        }
        // 如果有 mask 標志位，默認都是有的
        if (MASK) {
            var maskBytes = buf.slice(idx, idx + 4); //獲取掩碼數據
            idx += 4; //指針前移到真實數據段
            var payload = buf.slice(idx, idx + length); // 數據長度的單位是字節
            payload = unmask(maskBytes, payload); //解碼真實數據
        } else {
            payload = buf.slice(idx, idx + length);
        }

        this.buffer = buf.slice(idx + length); // 緩存 buffer
        // 有可能是分幀，需要拼接數據
        this.frames = Buffer.concat([this.frames, payload]); // 保存到 frames 中

        if (!FIN) {
            console.log(
                "server detect fragment, sizeof payload:",
                Buffer.byteLength(payload)
            );
        }

        if (FIN) {
            payload = this.frames.slice(0); // 獲取所有拼接完整的數據
            opcode = opcode || this.frameOpcode; // 如果是 0 ，則保持獲取之前保存的 code
            this.frames = Buffer.alloc(0); // 清空 frames
            this.frameOpcode = 0; // 清空 opcode
            this._handleFrame(opcode, payload); // 處理操作碼
        }

        return true; // 繼續處理
    }
    /**
     * 針對不同操作碼進行不同處理
     * @param 操作碼
     * @param 數據
     */
    _handleFrame(opcode, buffer) {
        var payload;
        switch (opcode) {
            case OPCODES.TEXT:
                payload = buffer.toString("utf8"); //如果是文本需要轉化為utf8的編碼
                this.emit("data", opcode, payload); //Buffer.toString()默認utf8 這里是故意指示的
                break;
            case OPCODES.BINARY: //二進制文件直接交付
                payload = buffer;
                this.emit("data", opcode, payload);
                break;
            case OPCODES.PING: // 發送 pong 做響應
                this._doSend(OPCODES.PONG, buffer);
                break;
            case OPCODES.PONG: //不做處理
                console.log("server receive pong");
                break;
            case OPCODES.CLOSE: // close有很多關閉碼
                let code, reason; // 用于獲取關閉碼和關閉原因
                if (buffer.length >= 2) {
                    code = buffer.readUInt16BE(0);
                    reason = buffer.toString("utf8", 2);
                }
                this.close(code, reason);
                this.emit("close", code, reason);
                break;
            default:
                this.close(1002, "unhandle opcode:" + opcode);
        }
    }
    // 這里可以針對 payload 的長度做分片
    _doSend(opcode, payload) {
        var len = Buffer.byteLength(payload);

        // 分片的距離邏輯
        var count = 0;
        while (len > MAX_FRAME_SIZE) {
            var framePayload = payload.slice(0, MAX_FRAME_SIZE);
            payload = payload.slice(MAX_FRAME_SIZE);
            this.socket.write(
                encodeMessage(
                    count > 0 ? OPCODES.CONTINUE : opcode,
                    framePayload,
                    false
                )
            ); //編碼后直接通過socket發送
            count++;
            len = Buffer.byteLength(payload);
        }

        this.socket.write(
            encodeMessage(count > 0 ? OPCODES.CONTINUE : opcode, payload)
        ); //編碼后直接通過socket發送
    }
}

module.exports = WebSocket;

運行測試代碼，打開瀏覽器訪問：http://localhost:3000/，在控制臺輸入：

var http = require('http');
var WebSocket = require('./websocket');
// HTTP服務器部分
var server = http.createServer(function(req, res) {
  res.end('websocket test\r\n');
});

console.log('starting...');

// Upgrade請求處理
server.on('upgrade', callback);

function callback(req, socket, upgradeHead) {
  var ws = new WebSocket(req, socket, upgradeHead);
  // ws.keepLive(); // 保持心跳連接，否則一般經過一定的時間沒有數據交互，瀏覽器端會主動關閉 ws 鏈接
  ws.on('data', function(opcode, payload) {
    console.log('receive data:', opcode, payload.length);
    ws.send('good job');
  });


  ws.on('close', function(code, reason) {
    console.log('close:', code, reason);
  });

}

server.listen(3000);