[[394780]]

前言：本文幾基于nodejs的ws模塊分析websocket的原理。

ws服務器邏輯由websocket-server.js的WebSocketServer類實現。該類初始化了一些參數后就執行以下代碼

if (this._server) { 
      // 給server注冊下面事件，返回一個注銷函數（用于注銷下面注冊的事件） 
      this._removeListeners = addListeners(this._server, { 
        // listen成功的回調 
        listening: this.emit.bind(this, 'listening'), 
        error: this.emit.bind(this, 'error'), 
        // 收到協議升級請求的回調 
        upgrade: (req, socket, head) => { 
          this.handleUpgrade(req, socket, head, (ws) => { 
            // 處理成功，觸發鏈接成功事件 
            this.emit('connection', ws, req); 
          }); 
        } 
      }); 

我們看到ws監聽了upgrade事件，當有websocket請求到來時就會執行handleUpgrade處理升級請求，升級成功后觸發connection事件。我們先看handleUpgrade。handleUpgrade邏輯不多，主要是處理和校驗升級請求的一些http頭。ws提供了一個校驗的鉤子。處理完http頭后，會調verifyClient校驗是否允許升級請求。如果成功則執行completeUpgrade。顧名思義，completeUpgrade是完成升級請求的函數，該函數返回同意協議升級并且設置一些http響應頭。另外還有一些重要的邏輯處理。

const ws = new WebSocket(null); 
// 設置管理socket的數據 
ws.setSocket(socket, head, this.options.maxPayload); 
// cb就是this.emit('connection', ws, req); 
cb(ws); 

我們看到這里新建了一個WebSocket對象并且調用了他的setSocket函數。我們來看看他做了什么。setSocket的邏輯非常多，我們慢慢分析。

數據接收者

class Receiver extends Writable {} 

我們看到數據接收者是一個可寫流。這就意味著我們可以往里面寫數據。

const receiver = new Receiver(); 
receiver.write('hello'); 

我們看一下這時候Receiver的邏輯。

_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); 
  } 

首先記錄當前數據的大小，然后把數據存起來，最后執行startLoop。

startLoop(cb) { 
    let err; 
    this._loop = true; 
 
    do { 
      switch (this._state) { 
        // 忽略其他case 
        case GET_DATA: 
          err = this.getData(cb); 
          break; 
        default: 
          // `INFLATING` 
          this._loop = false; 
          return; 
      } 
    } while (this._loop); 
 
    cb(err); 
  } 

我們知道websocket是基于tcp上層的應用層協議，所以我們收到數據時，需要解析出一個個數據包(粘包問題)，所以Receiver其實就是一個狀態機，每次收到數據的時候，都會根據當前的狀態進行狀態流轉。比如當前處于GET_DATA狀態，那么就會進行數據的處理。我們接著看一下數據處理的邏輯。

getData(cb) { 
    let data = EMPTY_BUFFER; 
    // 提取數據部分 
    if (this._payloadLength) { 
      data = this.consume(this._payloadLength); 
      if (this._masked) unmask(data, this._mask); 
    } 
    // 是控制報文則執行controlMessage 
    if (this._opcode > 0x07) return this.controlMessage(data); 
    // 做了壓縮，則先解壓 
    if (this._compressed) { 
      this._state = INFLATING; 
      this.decompress(data, cb); 
      return; 
    } 
    // 沒有壓縮則直接處理（先存到_fragments,然后執行dataMessage） 
    if (data.length) { 
      this._messageLength = this._totalPayloadLength; 
      this._fragments.push(data); 
    } 
 
    return this.dataMessage(); 
  } 

我們執行websocket協議定義了報文的類型，比如控制報文，數據報文。我們分別看一下這兩個的邏輯。

controlMessage(data) { 
    // 連接關閉 
    if (this._opcode === 0x08) { 
      this._loop = false; 
      if (data.length === 0) { 
        this.emit('conclude', 1005, ''); 
        this.end(); 
      } 
    } else if (this._opcode === 0x09) { 
      this.emit('ping', data); 
    } else { 
      this.emit('pong', data); 
    } 
    this._state = GET_INFO; 
  } 

我們看到控制報文包括三種(conclude、ping、pong)。而數據報文只有this.emit('message', data);一種。這個就是接收者的整體邏輯。

2 數據發送者

數據發送者是對websocket協議的封裝，當用戶調研數據發送者的send接口發送數據時，數據發送者會組裝成一個websocket協議的包再發送出去。

send(data, options, cb) { 
    const buf = toBuffer(data); 
    const perMessageDeflate = this._extensions[PerMessageDeflate.extensionName]; 
    let opcode = options.binary ? 2 : 1; 
    let rsv1 = options.compress; 
 
    if (this._firstFragment) { 
      this._firstFragment = false; 
      if (rsv1 && perMessageDeflate) { 
        rsv1 = buf.length >= perMessageDeflate._threshold; 
      } 
      this._compress = rsv1; 
    } else { 
      rsv1 = false; 
      opcode = 0; 
    } 
 
    if (options.fin) this._firstFragment = true; 
    // 需要壓縮 
    if (perMessageDeflate) { 
      const opts = { 
        fin: options.fin, 
        rsv1, 
        opcode, 
        mask: options.mask, 
        readOnly: toBuffer.readOnly 
      }; 
      // 正在壓縮，則排隊等待，否則執行壓縮 
      if (this._deflating) { 
        this.enqueue([this.dispatch, buf, this._compress, opts, cb]); 
      } else { 
        this.dispatch(buf, this._compress, opts, cb); 
      } 
    } else { 
      // 不需要壓縮，直接發送 
      this.sendFrame( 
        Sender.frame(buf, { 
          fin: options.fin, 
          rsv1: false, 
          opcode, 
          mask: options.mask, 
          readOnly: toBuffer.readOnly 
        }), 
        cb 
      ); 
    } 
  } 

send函數做了一些參數的處理后發送數據，但是如果需要壓縮的話，要壓縮后才能發送。數據處理完成后調用真正的發送函數

sendFrame(list, cb) { 
    if (list.length === 2) { 
      this._socket.cork(); 
      this._socket.write(list[0]); 
      this._socket.write(list[1], cb); 
      this._socket.uncork(); 
    } else { 
      this._socket.write(list[0], cb); 
    } 
  } 

了解了數據接收者和發送者的邏輯后，我們看一下websocket對象和setSocket函數做了什么事情，websocket對象本質是對TCP socket的封裝。它接收來自底層的數據，然后透傳給數據接收者，數據接收者處理完后，觸發websocket對應的對應的事件，比如message事件。發送數據的時候，websocket會調用數據發送者的接口，數據發送者組裝成websocket協議的數據包后再發送出去，架構如下圖所示。

接下來我們看看setSocket的邏輯

setSocket(socket, head, maxPayload) { 
    // 數據接收者，負責處理tcp上收到的數據(socket是tcp層的socket) 
    const receiver = new Receiver(...); 
    // 數據發送者，負責發送數據給對端 
    this._sender = new Sender(socket, this._extensions); 
    // 數據接收者，負責解析數據 
    this._receiver = receiver; 
    // net模塊的tcp socket 
    this._socket = socket; 
    // 關聯起來 
    receiver[kWebSocket] = this; 
    socket[kWebSocket] = this; 
    // 監聽接收者的事件，解析數據的時候會回調 
    receiver.on('conclude', receiverOnConclude); 
    // 下面兩個事件由Writable觸發 
    receiver.on('drain', receiverOnDrain); 
    receiver.on('error', receiverOnError); 
    receiver.on('message', receiverOnMessage); 
    receiver.on('ping', receiverOnPing); 
    receiver.on('pong', receiverOnPong); 
    // 清除定時器 
    socket.setTimeout(0); 
    // 關閉nagle算法 
    socket.setNoDelay(); 
    // 升級請求中，攜帶的http body，通常是空 
    if (head.length > 0) socket.unshift(head); 
    // 監聽tcp底層的事件 
    socket.on('close', socketOnClose); 
    socket.on('data', socketOnData); 
    socket.on('end', socketOnEnd); 
    socket.on('error', socketOnError); 
 
    this.readyState = WebSocket.OPEN; 
    this.emit('open'); 
  } 

我們看到里面監聽了各種事件，下面以data事件為例，看一下處理過程。當tcp socket收到數據的時候會執行socketOnData函數。

function socketOnData(chunk) { 
  // 會調用receiver里的_write函數，其實就是換成到receiver對象上，如果數據解析出錯，會觸發socket error事件 
  if (!this[kWebSocket]._receiver.write(chunk)) { 
    this.pause(); 
  } 
} 

socketOnData通過接收者的接口把數據傳給接收者，接收者會解析數據，然后觸發對應的事件，比如message。

receiver.on('message', receiverOnMessage); 
function receiverOnMessage(data) { 
  this[kWebSocket].emit('message', data); 
} 

然后ws的socket對象繼續往上層觸發message事件。this[kWebSocket]的值是ws提供的socket對象本身。架構圖如下。

這就是ws實現websocket協議的基本原理，具體細節可以參考源碼。