基于 Serverless 與 Websocket 的聊天工具實作

傳統業務實作 Websocket 并不難，然而函式計算基本上都是事件驅動，不支持長鏈接操作，如果將函式計算與 API 網關結合，是否可以有 Websocket 的實作方案呢？

API 網關觸發器實作 Websocket

WebSocket 協議是基于 TCP 的一種新的網路協議，它實作了瀏覽器與服務器全雙工 (full-duplex) 通信，即允許服務器主動發送資訊給客戶端，WebSocket 在服務端有資料推送需求時，可以主動發送資料至客戶端，而原有 HTTP 協議的服務端對于需推送的資料，僅能通過輪詢或 long poll 的方式來讓客戶端獲得，

由于云函式是無狀態且以觸發式運行，即在有事件到來時才會被觸發，因此，為了實作 WebSocket，云函式 SCF 與 API 網關相結合，通過 API 網關承接及保持與客戶端的連接，您可以認為云函式與 API 網關一起實作了服務端，當客戶端有訊息發出時，會先傳遞給 API 網關，再由 API 網關觸發云函式執行，當服務端云函式要向客戶端發送訊息時，會先由云函式將訊息 POST 到 API 網關的反向推送鏈接，再由 API 網關向客戶端完成訊息的推送，

具體的實作架構如下：

對于 WebSocket 的整個生命周期，主要由以下幾個事件組成：

• 連接建立：客戶端向服務端請求建立連接并完成連接建立；
• 資料上行：客戶端通過已經建立的連接向服務端發送資料；
• 資料下行：服務端通過已經建立的連接向客戶端發送資料；
• 客戶端斷開：客戶端要求斷開已經建立的連接；
• 服務端斷開：服務端要求斷開已經建立的連接，

對于 WebSocket 整個生命周期的事件，云函式和 API 網關的處理程序如下：

• 連接建立：客戶端與 API 網關建立 WebSocket 連接，API 網關將連接建立事件發送給 SCF；
• 資料上行：客戶端通過 WebSocket 發送資料，API 網關將資料轉發送給 SCF；
• 資料下行：SCF 通過向 API 網關指定的推送地址發送請求，API 網關收到后會將資料通過 WebSocket 發送給客戶端；
• 客戶端斷開：客戶端請求斷開連接，API 網關將連接斷開事件發送給 SCF；
• 服務端斷開：SCF 通過向 API 網關指定的推送地址發送斷開請求，API 網關收到后斷開 WebSocket 連接，

因此，云函式與 API 網關之間的互動，需要由 3 類云函式來承載：

• 注冊函式：在客戶端發起和 API 網關之間建立 WebSocket 連接時觸發該函式，通知 SCF WebSocket 連接的 secConnectionID，通常會在該函式記錄 secConnectionID 到持久存盤中，用于后續資料的反向推送；
• 清理函式：在客戶端主動發起 WebSocket 連接中斷請求時觸發該函式，通知 SCF 準備斷開連接的 secConnectionID，通常會在該函式清理持久存盤中記錄的該 secConnectionID；
• 傳輸函式：在客戶端通過 WebSocket 連接發送資料時觸發該函式，告知 SCF 連接的 secConnectionID 以及發送的資料，通常會在該函式處理業務資料，例如，是否將資料推送給持久存盤中的其他 secConnectionID，

Websocket 功能實作

根據騰訊云官網提供的該功能的整體架構圖：

這里我們可以使用物件存盤 COS 作為持久化的方案，當用戶建立鏈接存盤 ConnectionId 到 COS 中，當用戶斷開連接洗掉該鏈接 ID，

其中注冊函式：

# -*- coding: utf8 -*-
import os
from qcloud_cos_v5 import CosConfig
from qcloud_cos_v5 import CosS3Client

bucket = os.environ.get('bucket')
region = os.environ.get('region')
secret_id = os.environ.get('secret_id')
secret_key = os.environ.get('secret_key')
cosClient = CosS3Client(CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key))


def main_handler(event, context):
    print("event is %s" % event)

    connectionID = event['websocket']['secConnectionID']

    retmsg = {}
    retmsg['errNo'] = 0
    retmsg['errMsg'] = "ok"
    retmsg['websocket'] = {
        "action": "connecting",
        "secConnectionID": connectionID
    }

    cosClient.put_object(
        Bucket=bucket,
        Body='websocket'.encode("utf-8"),
        Key=str(connectionID),
        EnableMD5=False
    )

    return retmsg

傳輸函式：

# -*- coding: utf8 -*-
import os
import json
import requests
from qcloud_cos_v5 import CosConfig
from qcloud_cos_v5 import CosS3Client

bucket = os.environ.get('bucket')
region = os.environ.get('region')
secret_id = os.environ.get('secret_id')
secret_key = os.environ.get('secret_key')
cosClient = CosS3Client(CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key))

sendbackHost = os.environ.get("url")


def Get_ConnectionID_List():
    response = cosClient.list_objects(
        Bucket=bucket,
    )
    return [eve['Key'] for eve in response['Contents']]


def send(connectionID, data):
    retmsg = {}
    retmsg['websocket'] = {}
    retmsg['websocket']['action'] = "data send"
    retmsg['websocket']['secConnectionID'] = connectionID
    retmsg['websocket']['dataType'] = 'text'
    retmsg['websocket']['data'] = data
    requests.post(sendbackHost, json=retmsg)


def main_handler(event, context):
    print("event is %s" % event)

    connectionID_List = Get_ConnectionID_List()
    connectionID = event['websocket']['secConnectionID']
    count = len(connectionID_List)
    data = https://www.cnblogs.com/serverlesscloud/p/event['websocket']['data'] + "(===Online people:" + str(count) + "===)"
    for ID in connectionID_List:
        if ID != connectionID:
            send(ID, data)

    return "send success"

清理函式：

# -*- coding: utf8 -*-
import os
import requests
from qcloud_cos_v5 import CosConfig
from qcloud_cos_v5 import CosS3Client

bucket = os.environ.get('bucket')
region = os.environ.get('region')
secret_id = os.environ.get('secret_id')
secret_key = os.environ.get('secret_key')
cosClient = CosS3Client(CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key))

sendbackHost = os.environ.get("url")


def main_handler(event, context):
    print("event is %s" % event)

    connectionID = event['websocket']['secConnectionID']

    retmsg = {}
    retmsg['websocket'] = {}
    retmsg['websocket']['action'] = "closing"
    retmsg['websocket']['secConnectionID'] = connectionID
    requests.post(sendbackHost, json=retmsg)

    cosClient.delete_object(
        Bucket=bucket,
        Key=str(connectionID),
    )

    return event

Yaml 檔案如下：

Conf:
  component: "serverless-global"
  inputs:
    region: ap-guangzhou
    bucket: chat-cos-1256773370
    secret_id: 
    secret_key: 

myBucket:
  component: '@serverless/tencent-cos'
  inputs:
    bucket: ${Conf.bucket}
    region: ${Conf.region}

restApi:
  component: '@serverless/tencent-apigateway'
  inputs:
    region: ${Conf.region}
    protocols:
      - http
      - https
    serviceName: ChatDemo
    environment: release
    endpoints:
      - path: /
        method: GET
        protocol: WEBSOCKET
        serviceTimeout: 800
        function:
          transportFunctionName: ChatTrans
          registerFunctionName: ChatReg
          cleanupFunctionName: ChatClean


ChatReg:
  component: "@serverless/tencent-scf"
  inputs:
    name: ChatReg
    codeUri: ./code
    handler: reg.main_handler
    runtime: Python3.6
    region:  ${Conf.region}
    environment:
      variables:
        region: ${Conf.region}
        bucket: ${Conf.bucket}
        secret_id: ${Conf.secret_id}
        secret_key: ${Conf.secret_key}
        url: http://set-gwm9thyc.cb-guangzhou.apigateway.tencentyun.com/api-etj7lhtw

ChatTrans:
  component: "@serverless/tencent-scf"
  inputs:
    name: ChatTrans
    codeUri: ./code
    handler: trans.main_handler
    runtime: Python3.6
    region:  ${Conf.region}
    environment:
      variables:
        region: ${Conf.region}
        bucket: ${Conf.bucket}
        secret_id: ${Conf.secret_id}
        secret_key: ${Conf.secret_key}
        url: http://set-gwm9thyc.cb-guangzhou.apigateway.tencentyun.com/api-etj7lhtw

ChatClean:
  component: "@serverless/tencent-scf"
  inputs:
    name: ChatClean
    codeUri: ./code
    handler: clean.main_handler
    runtime: Python3.6
    region:  ${Conf.region}
    environment:
      variables:
        region: ${Conf.region}
        bucket: ${Conf.bucket}
        secret_id: ${Conf.secret_id}
        secret_key: ${Conf.secret_key}
        url: http://set-gwm9thyc.cb-guangzhou.apigateway.tencentyun.com/api-etj7lhtw

注意，這里需要先部署 API 網關，當部署完成，獲得回推地址，將回推地址以 url 的形式寫入到對應函式的環境變數中：

理論上應該是可以通過 ${restApi.url[0].internalDomain} 自動獲得到 url 的，但是我并沒有成功獲得到這個 url，只能先部署 API 網關，獲得到這個地址之后，再重新部署，

部署完成之后，我們可以撰寫 HTML 代碼，實作可視化的 Websocket Client，其核心的 JavaScript 代碼為：

window.onload = function () {
    var conn;
    var msg = document.getElementById("msg");
    var log = document.getElementById("log");

    function appendLog(item) {
        var doScroll = log.scrollTop === log.scrollHeight - log.clientHeight;
        log.appendChild(item);
        if (doScroll) {
            log.scrollTop = log.scrollHeight - log.clientHeight;
        }
    }

    document.getElementById("form").onsubmit = function () {
        if (!conn) {
            return false;
        }
        if (!msg.value) {
            return false;
        }
        conn.send(msg.value);
        //msg.value = https://www.cnblogs.com/serverlesscloud/p/"";
		
		var item = document.createElement("div");
		item.innerText = "發送↑:";
		appendLog(item);
		
		var item = document.createElement("div");
		item.innerText = msg.value;
		appendLog(item);
		
        return false;
    };

    if (window["WebSocket"]) {
        //替換為websocket連接地址
        conn = new WebSocket("ws://service-01era6ni-1256773370.gz.apigw.tencentcs.com/release/");
        conn.onclose = function (evt) {
            var item = document.createElement("div");
            item.innerHTML = "Connection closed.";
            appendLog(item);
        };
        conn.onmessage = function (evt) {
			var item = document.createElement("div");
			item.innerText = "接收↓:";
			appendLog(item);
		
            var messages = evt.data.split('\n');
            for (var i = 0; i < messages.length; i++) {
                var item = document.createElement("div");
                item.innerText = messages[i];
                appendLog(item);
            }
        };
    } else {
        var item = document.createElement("div");
        item.innerHTML = "<b>Your browser does not support WebSockets.</b>";
        appendLog(item);
    }
};

完成之后，我們打開兩個頁面，進行測驗：

總結

通過云函式 + API 網關進行 Websocket 的實踐，絕對不僅僅是一個聊天工具這么簡單，它可以用在很多方面，例如通過 Websocket 進行實時日志系統的制作等，

單獨的函式計算，僅僅是一個計算平臺，只有和周邊的 BaaS 結合，才能展示出 Serverless 架構的價值和真正的能力，這也是為什么很多人說 Serverless=FaaS+BaaS 的一個原因，

期待更多小伙伴，可以通過 Serverless 架構，創造出更多有趣的應用，

Serverless Framework 30 天試用計劃

我們誠邀您來體驗最便捷的 Serverless 開發和部署方式，在試用期內，相關聯的產品及服務均提供免費資源和專業的技術支持，幫助您的業務快速、便捷地實作 Serverless！

詳情可查閱：Serverless Framework 試用計劃

One More Thing

3 秒你能做什么？喝一口水，看一封郵件，還是 —— 部署一個完整的 Serverless 應用？

復制鏈接至 PC 瀏覽器訪問：https://serverless.cloud.tencent.com/deploy/express

3 秒極速部署，立即體驗史上最快的 Serverless HTTP 實戰開發！

傳送門：

• GitHub: github.com/serverless
• 官網：serverless.com

歡迎訪問：Serverless 中文網，您可以在 最佳實踐 里體驗更多關于 Serverless 應用的開發！

推薦閱讀：《Serverless 架構：從原理、設計到專案實戰》

標籤：其他

上一篇：使用容器化塊存盤OpenEBS在K3s中實作持久化存盤

下一篇：【趙強老師】管理Docker鏡像