透過 WebSocket 串流傳送 Pub/Sub 訊息

簡介

越來越多應用程式採用事件驅動模型，因此前端應用程式必須能夠輕鬆地與訊息服務建立連結，而這些服務正是這類架構的基石。

您可以透過多種方式將資料串流至網路瀏覽器用戶端，其中最常見的是 WebSockets。本教學課程將逐步說明如何安裝程序，訂閱發布至 Pub/Sub 主題的訊息串流，並透過網路伺服器將這些訊息傳送至透過 WebSocket 連線的用戶端。

在本教學課程中，您將使用 NYC Taxi Tycoon Google Dataflow CodeLab 中使用的公開 Pub/Sub 主題。本主題提供模擬計程車遙測資料的即時串流，這些資料是以紐約市的歷來乘車資料為基礎，取自計程車暨禮車管理局的行程記錄資料集。

架構

下圖顯示您在本教學課程中建構的教學課程架構。

這張圖表顯示訊息發布者位於含有 Compute Engine 資源的專案外部，發布者會將訊息傳送至 Pub/Sub 主題。Compute Engine 執行個體會透過 WebSocket，將訊息提供給執行 HTML5 和 JavaScript 儀表板的瀏覽器。

本教學課程會結合使用下列工具，在 Pub/Sub 和 WebSocket 之間建立橋樑：

• pulltop 是您在本教學課程中安裝的 Node.js 程式。這項工具會訂閱 Pub/Sub 主題，並將收到的訊息串流至標準輸出。
• websocketd 是一個小型指令列工具，可包裝現有的指令列介面程式，並允許使用 WebSocket 存取。

結合 pulltop 和 websocketd，即可將從 Pub/Sub 主題收到的訊息，透過 WebSocket 串流至瀏覽器。

調整 Pub/Sub 主題輸送量

紐約市計程車業大亨公開 Pub/Sub 主題每秒會產生 2000 到 2500 個模擬計程車乘車更新，每秒最多可達 8 MB 以上。如果 Pub/Sub 偵測到未確認訊息的佇列不斷增加，就會自動減緩訂閱端的訊息速率。因此，不同工作站、網路連線和前端處理程式碼的訊息速率變異性可能較高。

有效處理瀏覽器訊息

由於 WebSocket 串流傳送的訊息量很大，因此您需要謹慎編寫處理這個串流的前端程式碼。舉例來說，您可能會為每則訊息動態建立 HTML 元素。但在預期訊息速率下，為每則訊息更新頁面可能會鎖定瀏覽器視窗。動態建立 HTML 元素導致頻繁分配記憶體，也會延長垃圾收集時間，進而降低使用者體驗。簡而言之，您不希望為每秒抵達的約 2000 則訊息呼叫 document.createElement()。

本教學課程採用的方法如下，可管理這類密集的訊息串流：

• 即時計算及持續更新一組串流指標，並以匯總值的形式顯示觀察到的訊息相關資訊。
• 使用以瀏覽器為基礎的資訊主頁，以預先定義的時間表顯示少量個別訊息，並即時顯示卸貨和取貨事件。

下圖顯示在本教學課程中建立的資訊主頁。

圖中顯示，在每秒近 2100 則訊息的速率下，最後一則訊息的延遲時間為 24 毫秒。如果處理每則訊息的重要程式碼路徑未及時完成，最後一則訊息的延遲時間會增加，每秒觀察到的訊息數量也會減少。系統會使用 JavaScript setInterval API 進行騎乘取樣，每三秒循環一次，避免前端在生命週期內建立大量 DOM 元素。(無論如何，每秒超過 10 個的事件絕大多數都無法實際觀察到)。

資訊主頁會在串流中途開始處理事件，因此除非先前已偵測到行程，否則資訊主頁會將進行中的行程視為新行程。程式碼會使用關聯陣列儲存每個觀察到的行程，並依 ride_id 值建立索引，且會在乘客下車時移除特定行程的參照。除非行程先前已觀察到 (「enroute」的情況)，否則處於「enroute」或「pickup」狀態的行程會將參照新增至該陣列。

安裝及設定 WebSocket 伺服器

首先，請建立 Compute Engine 執行個體，做為 WebSocket 伺服器。建立執行個體後，請在執行個體上安裝稍後會用到的工具。

1. 在 Cloud Shell 中，設定預設的 Compute Engine 區域。 以下範例顯示 us-central1-a，但您可以使用任何想要的區域。

   gcloud config set compute/zone us-central1-a
   

2. 在預設區域中建立名為 websocket-server 的 Compute Engine 執行個體：

   gcloud compute instances create websocket-server --tags wss
   

3. 新增防火牆規則，允許通訊埠 8000 的 TCP 流量傳送至標記為 wss 的任何執行個體：

   gcloud compute firewall-rules create websocket \
       --direction=IN \
       --allow=tcp:8000 \
       --target-tags=wss
   

4. 如果您使用現有專案，請確認 TCP 通訊埠 22 已開啟，允許 SSH 連線至執行個體。

   預設網路中的default-allow-ssh防火牆規則預設為啟用。不過，如果您或管理員在現有專案中移除了預設規則，TCP 通訊埠 22 可能不會開啟。(如果您是為了這個教學課程建立新專案，系統會預設啟用這項規則，因此您不必採取任何行動)。

   新增防火牆規則，允許通訊埠 22 的 TCP 流量傳送至標記為 wss 的任何執行個體：

   gcloud compute firewall-rules create wss-ssh \
       --direction=IN \
       --allow=tcp:22 \
       --target-tags=wss
   

5. 使用 SSH 連線至執行個體：

   gcloud compute ssh websocket-server
   

6. 在執行個體的終端機指令中，將帳戶切換為 root，以便安裝軟體：

   sudo -s
   

7. 安裝 git 和 unzip 工具：

   apt-get install -y unzip git
   

8. 在執行個體上安裝 websocketd 二進位檔：

   cd /var/tmp/
   wget \
   https://github.com/joewalnes/websocketd/releases/download/v0.3.0/websocketd-0.3.0-linux_386.zip
   unzip websocketd-0.3.0-linux_386.zip
   mv websocketd /usr/bin
   

安裝 Node.js 和教學課程程式碼

1. 在執行個體的終端機中安裝 Node.js：

   curl -sL https://deb.nodesource.com/setup_10.x | bash -
   apt-get install -y nodejs
   

2. 下載教學課程來源存放區：

   exit
   cd ~
   git clone https://github.com/GoogleCloudPlatform/solutions-pubsub-websockets.git
   

3. 變更 pulltop 的權限，允許執行：

   cd solutions-pubsub-websockets
   chmod 755 pulltop/pulltop.js
   

4. 安裝 pulltop 依附元件：

   cd pulltop
   npm install
   sudo npm link
   

測試 pulltop 是否可以讀取訊息

1. 在執行個體上，針對公開主題執行 pulltop：

   pulltop projects/pubsub-public-data/topics/taxirides-realtime
   

   如果 pulltop 正常運作，您會看到如下所示的結果串流：

   {"ride_id":"9729a68d-fcde-484b-bc32-bf29f5188628","point_idx":328,"latitude"
   :40.757360000000006,"longitude":-73.98228,"timestamp":"2019-03-22T20:03:51.6
   593-04:00","meter_reading":11.069151,"meter_increment":0.033747412,"ride_stat
   us":"enroute","passenger_count":1}

2. 按下 Ctrl+C 即可停止串流。

建立傳送至 websocketd 的訊息流程

現在您已確認 pulltop 可以讀取 Pub/Sub 主題，可以啟動 websocketd 程序，開始將訊息傳送至瀏覽器。

將主題訊息擷取到本機檔案

在本教學課程中，您會擷取從 pulltop 取得的訊息串流，並將其寫入本機檔案。將訊息流量擷取到本機檔案會增加儲存空間需求，但也會將 websocketd 程序作業與串流 Pub/Sub 主題訊息分離。在本地擷取資訊可讓您暫時停止 Pub/Sub 串流 (可能為了調整流量控管參數)，但不會強制重設目前連線的 WebSocket 用戶端。訊息串流重新建立後， websocketd會自動繼續將訊息串流傳送給用戶端。

1. 在執行個體上，針對公開主題執行 pulltop，並將訊息輸出內容重新導向至本機 taxi.json 檔案。nohup 指令會指示 OS 在您登出或關閉終端機時，讓 pulltop 程序保持執行狀態。

   nohup pulltop \
     projects/pubsub-public-data/topics/taxirides-realtime > \
     /var/tmp/taxi.json &
   

2. 確認 JSON 訊息是否正在寫入檔案：

   tail /var/tmp/taxi.json
   

   如果訊息寫入 taxi.json 檔案，輸出內容會類似下列內容：

   {"ride_id":"9729a68d-fcde-484b-bc32-bf29f5188628","point_idx":328,"latitude"
   :40.757360000000006,"longitude":-73.98228,"timestamp":"2019-03-22T20:03:51.6
   593-04:00","meter_reading":11.069151,"meter_increment":0.033747412,"ride_sta
   tus":"enroute","passenger_count":1}

3. 切換至應用程式的網頁資料夾：

   cd ../web
   

4. 啟動 websocketd，開始使用 WebSockets 串流處理本機檔案內容：

   nohup websocketd --port=8000 --staticdir=. tail -f /var/tmp/taxi.json &
   

   這會在背景執行 websocketd 指令。websocketd 工具會取用 tail 指令的輸出內容，並將每個元素串流為 WebSocket 訊息。

5. 檢查 nohup.out 的內容，確認伺服器是否正確啟動：

   tail nohup.out
   

   如果一切運作正常，輸出內容會如下所示：

   Mon, 25 Mar 2019 14:03:53 -0400 | INFO   | server     |  | Serving using application   : /usr/bin/tail -f /var/tmp/taxi.json
   Mon, 25 Mar 2019 14:03:53 -0400 | INFO   | server     |  | Serving static content from : .
   

以視覺化方式呈現訊息

發布至 Pub/Sub 主題的個別行程訊息結構如下：

{
  "ride_id": "562127d7-acc4-4af9-8fdd-4eedd92b6e69",
  "point_idx": 248,
  "latitude": 40.74644000000001,
  "longitude": -73.97144,
  "timestamp": "2019-03-24T00:46:08.49094-04:00",
  "meter_reading": 8.40615,
  "meter_increment": 0.033895764,
  "ride_status": "enroute",
  "passenger_count": 1
}

根據這些值，您可以計算資訊主頁標題的幾項指標。 系統會針對每個進站行程事件執行一次計算。可能的值包括：

• 最後一則訊息的延遲時間。上次觀察到的行程事件時間戳記與目前時間之間的時間差 (以秒為單位)，目前時間是從代管網頁瀏覽器的系統時鐘衍生而來。
• 進行中的行程。目前進行中的行程數量。這個數字可能會快速增加，但如果觀察到 ride_status 值為 dropoff，這個數字就會減少。
• 訊息費率。每秒處理的平均行程事件數。
• 總計量金額。所有進行中行程的計費表總和。 隨著行程結束，這個數字會隨之減少。
• 乘客總人數。所有行程的乘客人數。這個數字會隨著行程完成而減少。
• 每趟行程的平均乘客人數。總搭乘次數除以總乘客人數。
• 每位乘客的平均計費金額。總計費金額除以乘客總人數。

除了指標和個別行程樣本外，當乘客上車或下車時，資訊主頁的行程樣本格狀檢視畫面上方會顯示快訊通知。

1. 取得目前執行個體的外部 IP 位址：

   curl -H "Metadata-Flavor: Google" http://metadata/computeMetadata/v1/instance/network-interfaces/0/access-configs/0/external-ip; echo
   
   

2. 複製 IP 位址，

3. 在本機上開啟新的網路瀏覽器，然後輸入下列網址：

   http://$ip-address:8000。

   您會看到顯示本教學課程資訊主頁的頁面：

   

4. 按一下頂端的計程車圖示，開啟串流連線並開始處理訊息。

   系統會每三秒算繪九個有效行程的樣本，以視覺化呈現個別行程：

   

   你可以隨時點選計程車圖示，啟動或停止 WebSocket 串流。如果 WebSocket 連線中斷，圖示會變成紅色，且系統會停止更新指標和個別行程。如要重新連線，請再次點選計程車圖示。

成效

下圖顯示 Chrome 開發人員工具的效能監控器，當時瀏覽器分頁每秒處理約 2100 則訊息。

訊息傳送延遲時間約為 30 毫秒，CPU 使用率平均約為 80%。記憶體使用率至少為 29 MB，總共分配 57 MB，且可自由增加或減少。