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

本教學課程說明前端應用程式 (在本例中為網頁) 如何在使用Google Cloud時處理大量傳入資料。本教學課程說明大量串流的幾項挑戰，本教學課程提供範例應用程式，說明如何使用 WebSockets 顯示發布至 Pub/Sub 主題的密集訊息串流，並及時處理這些訊息，維持高效能的前端。

本教學課程適用於熟悉透過 HTTP 進行瀏覽器與伺服器通訊，以及使用 HTML、CSS 和 JavaScript 編寫前端應用程式的開發人員。本教學課程假設您曾經使用過Google Cloud，並熟悉 Linux 指令列工具。

目標

• 建立及設定虛擬機器 (VM) 執行個體，並加入必要元件，將 Pub/Sub 訂閱項目的酬載串流至瀏覽器用戶端。
• 在 VM 上設定程序，訂閱 Pub/Sub 主題，並將個別訊息輸出至記錄檔。
• 安裝網路伺服器，提供靜態內容，並將殼層指令輸出內容串流至 WebSocket 用戶端。
• 使用 HTML、CSS 和 JavaScript，在瀏覽器中顯示 WebSocket 串流匯總和個別訊息樣本。

費用

在本文件中，您會使用下列 Google Cloud的計費元件：

• Compute Engine
• Pub/Sub

您可以使用 Pricing Calculator，根據預測用量估算費用。

初次使用 Google Cloud 的使用者可能符合免費試用期資格。

事前準備

Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

• Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
• Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

Go to project selector

Verify that billing is enabled for your Google Cloud project.

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

• Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
• Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

Go to project selector

Verify that billing is enabled for your Google Cloud project.

1. 開啟 Cloud Shell，執行本教學課程中列出的指令。

   前往 Cloud Shell

   您將從 Cloud Shell 執行本教學課程中所有的終端機指令。

2. 啟用 Compute Engine API 和 Pub/Sub API：

   gcloud services enable compute pubsub

完成本教學課程後，您可以刪除建立的資源以避免繼續計費。詳情請參閱清除所用資源一節。

簡介

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

您可以透過多種方式將資料串流至網路瀏覽器用戶端，其中最常見的是 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 進行行程取樣，並將 API 設為每三秒循環一次，避免前端在生命週期內建立大量 DOM 元素。(無論如何，每秒超過 10 個的速率，絕大多數都實際上無法觀察到)。

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

安裝及設定 WebSocket 伺服器

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

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，開始使用 WebSocket 串流處理本機檔案的內容：

   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，且可自由增加和減少。

清除所用資源

移除防火牆規則

如果您使用現有專案進行本教學課程，可以移除您建立的防火牆規則。建議您盡量減少開放的連接埠。

1. 刪除您建立的防火牆規則，允許通訊埠 8000 上的 TCP：

   gcloud compute firewall-rules delete websocket
   

2. 如果您也建立了防火牆規則來允許 SSH 連線，請刪除允許通訊埠 22 上 TCP 的防火牆規則：

   gcloud compute firewall-rules delete wss-ssh
   

刪除專案

如果不想再使用這個專案，可以刪除專案。

注意：刪除專案會造成以下影響：
• 專案中的所有內容都會遭到刪除。如果使用現有專案來進行本文中的任務，刪除專案將一併移除當中已完成的其他任務'。
• 自訂專案 ID 會消失。當您之前建立這個專案時，可能建立了想要在日後使用的自訂專案 ID。如要保留使用該專案 ID 的網址 (例如 appspot.com 網址)，請刪除在該專案中選取的資源，而不是刪除整個專案。
1. In the Google Cloud console, go to the Manage resources page.

   Go to Manage resources

2. In the project list, select the project that you want to delete, and then click Delete.
3. In the dialog, type the project ID, and then click Shut down to delete the project.

後續步驟

• 進一步瞭解 Pub/Sub 和 WebSockets 通訊協定
• 將 Google 地圖平台 API 金鑰新增至 cabdash.js，即可根據上車和下車地點進行地理位置查詢。
• 探索 Google Cloud 的參考架構、圖表和最佳做法。歡迎瀏覽我們的雲端架構中心。