互動直播中的前端技術——即時通訊

阿新 • • 發佈：2020-08-17

前言

在疫情期間，上班族開啟了遠端辦公，體驗了各種遠端辦公軟體。老師做起了主播，學生們感受到了被釘釘支配的恐懼，歌手們開啟了線上演唱會，許多綜藝節目也變成了線上直播。在這全民互動直播的時期，我們來聊聊互動直播中的即時通訊技術在前端中的使用。

即時通訊技術

即時通訊（Instant Messaging，簡稱IM）是一個實時通訊系統，允許兩人或多人使用網路實時的傳遞文字訊息、檔案、語音與視訊交流。如何來實現呢，通常我們會使用伺服器推送技術來實現。常見的有以下幾種實現方式。

輪詢(polling)

這是一種我們幾乎都用到過的的技術實現方案。客戶端和伺服器之間會一直進行連線，每隔一段時間就詢問一次。前端通常採取setInterval或者setTimeout去不斷的請求伺服器資料。

缺點：輪詢時間通常是死的，太長就不是很實時，太短增加伺服器端的負擔。不斷的去請求沒有意義的更新的資料也是一種浪費伺服器資源的做法。

長輪詢(long-polling)

客戶端傳送一個請求到服務端，如果服務端沒有新的資料，就保持住這個連線直到有資料。一旦服務端有了資料（訊息）給客戶端，它就使用這個連線傳送資料給客戶端。接著連線關閉。

缺點：佔較多的記憶體資源與請求數。

iframe流

iframe流就是在瀏覽器中動態載入一個iframe, 讓它的地址指向請求的伺服器的指定地址（就是向伺服器傳送了一個http請求），然後在瀏覽器端建立一個處理資料的函式，在服務端通過iframe與瀏覽器的長連線定時輸出資料給客戶端，iframe頁面接收到這個資料就會將它解析成程式碼並傳資料給父頁面從而達到即時通訊的目的。

缺點：相容性與使用者體驗不好。伺服器維護一個長連線會增加開銷。一些瀏覽器的的位址列圖示會一直轉菊花。

Server-sent Events(sse)

sse與長輪詢機制類似，區別是每個連線不只傳送一個訊息。客戶端傳送一個請求，服務端保持這個連線直到有新訊息傳送回客戶端，仍然保持著連線，這樣連線就可以訊息的再次傳送，由伺服器單向傳送給客戶端。

缺點：相容性不好（IE，Edge不支援）；伺服器只能單向推送資料到客戶端。

WebSocket

HTML5 WebSocket規範定義了一種API，使Web頁面能夠使用WebSocket協議與遠端主機進行雙向通訊。與輪詢和長輪詢相比，巨大減少了不必要的網路流量和等待時間。

WebSocket屬於應用層協議。它基於TCP傳輸協議，並複用HTTP的握手通道。但不是基於HTTP協議的，只是在建立連線之前要藉助一下HTTP，然後在第一次握手是升級協議為ws或者wss。

缺點：開發成本高，需要額外做重連保活。

在互動直播場景下，由於本身的實時性要求高，服務端與客戶端需要頻繁雙向通訊，因此與它十分契合。

搭建自己的IM系統

上面簡單的概述了下即時通訊的實現技術，接下來我們就聊聊如何實現自己的IM系統。

從零開始搭建IM系統還是一件比較複雜與繁瑣的事情。自己搭建推薦基於socket.io來實現。socket.io對即時通訊的封裝已經很不錯了，是一個比較成熟的庫，對不同瀏覽器做了相容，提供了各端的方案包括服務端，我們不用關心底層是用那種技術實現進行資料的通訊，當然在現代瀏覽器種基本上是基於WebSocket來實現的。市面上也有不少IM雲服務平臺，比如雲信，藉助第三方的服務也可以快速整合。下面就介紹下前端怎麼基於socket.io整合開發。

基礎的搭建

服務端整合socket.io(有java版本的)，服務端即成可以參考下這裡，客戶端使用socket.io-client整合。

參考socket.io官方api，訂閱生命週期與事件，通過訂閱的方式或來實現基礎功能。在回撥函式執行解析包裝等邏輯，最終拋給上層業務使用。

import io from 'socket.io-client';
import EventEmitter from 'EventEmitter';
class Ws extends EventEmitter {
    constructor (options) {
        super();
        //...
        this.init();
    }
    init () {
        const socket  = this.link = io('wss://x.x.x.x');
        socket.on('connect', this.onConnect.bind(this));
        socket.on('message', this.onMessage.bind(this));
        socket.on('disconnect', this.onDisconnect.bind.(this);
        socket.on('someEvent', this.onSomeEvent.bind(this));
    }
    onMessage(msg) {
        const data = this.parseData(msg);
        // ...
        this.$emit('message', data);
    }
}

訊息收發

與伺服器或者其他客戶端進行訊息通訊時通常會基於業務約定協議來封裝解析訊息。由於都是非同步行為，需要有唯一標識來處理訊息回撥。這裡用自增seq來標記。

傳送訊息

class Ws extends EventEmitter {
    seq = 0;
    cmdTasksMap = {};
    // ...
    sendCmd(cmd, params) {
        return new Promise((resolve, reject) => {
            this.cmdTasksMap[this.seq] = {
                resolve,
                reject
            };
            const data = genPacket(cmd, params, this.seq++);
            this.link.send({ data });
        });
    }
}

接受訊息

class Ws extends EventEmitter {
    // ...
    onMessage(packet) {
        const data = parsePacket(packet);
        if (data.seq) {
            const cmdTask = this.cmdTasksMap[data.seq];
            if (cmdTask) {
                if (data.body.code === 200) {
                    cmdTask.resolve(data.body);
                } else {
                    cmdTask.reject(data.body);
                }
                delete this.cmdTasksMap[data.seq];
            }
        }
    }
}

生產環境中優化

上文只介紹了基礎功能的簡單封裝，在生產環境中使用，還需要對考慮很多因素，尤其是在互動直播場景中，禮物展示，麥序（進行語音通話互動的順序），聊天，群聊等都強依賴長連結的穩定性，下面就介紹一些兜底與優化措施。

連線保持

為了穩定建立長連結與保持長連結。採用了以下幾個手段：

超時處理
心跳包
重連退避機制

超時處理

在實際使用中，並不一定每次傳送訊息都服務端都有響應，可能在客戶端已經出現異常了，我們與服務端的通訊方式都是一問一答。基於這一點，我們可以增加超時邏輯來判斷是否是傳送成功。然後基於回撥上層進行有友好提示，進入異常處理。接下來就進一步改造傳送邏輯。

class Ws extends EventEmitter {
    // ...
    sendCmd(cmd, params) {
        return new Promise((resolve, reject) => {
            this.cmdTasksMap[this.seq] = {
                resolve,
                reject
            };
            // 加個定時器
            this.timeMap[this.seq] = setTimeout(() => {
                const err = new newTimeoutError(this.seq);
                reject({ ...err });
            }, CMDTIMEOUT);

            const data = genPacket(cmd, params, this.seq++);
            this.link.send({ data });
        });
    }
    onMessage(packet) {
        const data = parsePacket(packet);
        if (data.seq) {
            const cmdTask = this.cmdTasksMap[data.seq];
            if (cmdTask) {
                clearTimeout(this.timeMap[this.seq]);
                delete this.timeMap[this.seq];
                if (data.body.code === 200) {
                    cmdTask.resolve(data.body);
                } else {
                    cmdTask.reject(data.body);
                }
                delete this.cmdTasksMap[data.seq];
            }
        }
    }
}

心跳包

心跳包: 心跳包就是在客戶端和伺服器間定時通知對方自己狀態的一個自己定義的命令字，按照一定的時間間隔傳送，類似於心跳，所以叫做心跳包。

心跳包是檢查長連結存活的關鍵手段，在web端我們通過心跳包是否超時來判斷。TCP中已有keepalive選項，為什麼要在應用層加入心跳包機制？

tcp keepalive檢查連線是否存活
應用keepalive檢測應用是否正常可響應

舉個栗子: 服務端死鎖，無法處理任何業務請求。但是作業系統仍然可以響應網路層keepalive包。所以我們通常使用空內容的心跳包並設定合適的傳送頻率與超時時間來作為連線的保持的判斷。

如果服務端只認心跳包作為連線存在判斷，那就在連線建立後定時發心跳就行。如果以收到包為判斷存活，那就在每次收到訊息重置並起個定時器傳送心跳包。

class Ws extends EventEmitter {
    // ...
     onMessage(packet) {
        const data = parsePacket(packet);
        if (data.seq) {
            const cmdTask = this.cmdTasksMap[data.seq];
            if (cmdTask) {
                clearTimeout(this.timeMap[this.seq]);
                if (data.body.code === 200) {
                    cmdTask.resolve(data.body);
                } else {
                    cmdTask.reject(data.body);
                }
                delete this.cmdTasksMap[data.seq];
            }
        }
        this.startHeartBeat();
    }
    startHeartBeat() {
        if (this.heartBeatTimer) {
            clearTimeout(this.heartBeatTimer);
            this.heartBeatTimer = null;
        }
        this.heartBeatTimer = setTimeout(() => {
            // 在sendCmd中指定heartbeat型別seq為0，讓業務包連續編號
            this.sendCmd('heartbeat').then(() => {
                // 傳送成功了就不管
            }).catch((e) => {
                this.heartBeatError(e);
            });
        }, HEARTBEATINTERVAL);
    }
}

重連退避機制

連不上了，重連，還連不上，重連，又連不上，重連。重連是一個保活的手段，但總不能一直重連吧，因此我們要用合理策去重連。

通常服務端會提供lbs（Location Based Services，LBS）介面，來提供最優節點，我們端上要做便是快取這些地址並設定端上的重連退避機制。按級別次數通常會做以下處理。

重連（超時<X次）
換連線地址重連 (超時>=X次)
重新獲取連線地址(X<MAX)
上層處理（超過MAX）

在重連X次後選擇換地址，在一個地址失敗後，選擇重新去拿地址再去迴圈嘗試。具體的嘗試次數根據實際業務來定。當然在一次又一次失敗中做好異常上報，以便於分析解決問題。

接受訊息優化

在高併發的場景下尤其是聊天室場景，我們要做一定的訊息合併與緩衝，來避免過多的UI繪製與應用阻塞。

因此要約定好解析協議，服務端與客戶端都做訊息合併，並設定訊息緩衝。示例如下：

Fn.startMsgFlushTimer = function () {
    this.msgFlushTimer = setTimeout(() => {
    const msgs = this.msgBuffer.splice(0, BUFFERSIZE);
    // 回撥訊息通知
    this.onmsgs(msgs);
    if (!this.msgBuffer.length) {
      this.msgFlushTimer = null;
    } else {
      this.startMsgFlushTimer();
    }
  }, MSGBUFFERINTERVAL);
};

流量優化

持久化儲存

在單聊場景中每次都同步全量的會話，歷史訊息等這是一個很大的代價。此外關閉web也是一種比較容易的操作(基本上就需要重新同步一次)。如果我們用增量的方式去同步就可以減少很多流量。實現增量同步自然想到了web儲存。

常用web儲存cookie，localStorage，sessionStorage不太能滿足我們持久化的場景，然而html5的indexedDB正常好滿足我們的需求。IndexedDB 內部採用物件倉庫（object store）存放資料。所有型別的資料都可以直接存入，包括JavaScript物件。indexedDB的api直接用可能會比較難受，可以使用Dexie.js，db.js這些二次封裝的庫來實現業務的資料層。

在滿足持久化儲存後, 我們便可以用時間戳，來進行增量同步，在收到訊息通知時，儲存到web資料庫。上層操作獲取資料，優先從資料庫獲取資料，避免總是高頻率、高資料量的與伺服器通訊。當然敏感性資訊不要存在資料庫或者增加點破解難度，畢竟所有web本地儲存都是能看到的。此外注意下儲存大小還是有限制的，每種瀏覽器可能不一樣，但是遠大於其他Web本地儲存了，只要該放雲端的資料放雲端(比如雲訊息)，不會有太大問題。

在編碼實現上，由於處理訊息通知都是非同步操作，要維護一個佇列保證入庫時序。此外要做好降級方案。

減少連線數

在Web桌面端的互動直播場景，同一種頁面開啟了多個tab訪問應該是很常見的。業務上也會有多端互踢操作，但是對Web場景如果只能一個頁面能進行互動那肯定是不行的，一不小心就不知道切到哪個tab上去了。所以通常會設定一個多端線上的最大數，超過了就踢。因而一個瀏覽器建立7，8個長連結是一件很尋常的事情，對於服務端資源也是一種極大的浪費。

Web Worker可以為Web內容在後臺執行緒中執行指令碼提供了一種簡單的方法，執行緒可以執行任務而不干擾使用者介面。並且可以將訊息傳送到建立它的JavaScript程式碼, 通過將訊息釋出到該程式碼指定的事件處理程式（反之亦然）。雖然Web Worker中不能使用DOM API，但是XHR，WebSocket這些通訊API並沒有限制（而且可以操作本地儲存）。因此我們可以通過SharedWorker API建立一個執行指定指令碼來共享web worker來實現多個tab之前的通訊複用，來達到減少連線數的目的。在相容性要求不那麼高的場景可以嘗試一下。

小結

本文介紹了互動直播中的即時通訊技術的在前端中應用，並分享了自己在工作開發中的一些經驗，希望對您有所幫助，歡迎探討。