Streaming von Pub/Sub-Nachrichten über WebSockets

Einleitung

Immer mehr Anwendungen arbeiten nach ereignisgesteuerten Modellen. Daher ist es wichtig, dass Frontend-Anwendungen einfach und reibungslos Verbindungen zu den Messaging-Diensten herstellen können, die diesen Architekturen zugrunde liegen.

Es gibt verschiedene Möglichkeiten für das Streaming von Daten an Webbrowser-Clients. Am häufigsten werden WebSockets verwendet. Diese Anleitung führt Sie durch die Installation eines Prozesses, mit dem ein Nachrichtenstrom abonniert, in einem Pub/Sub-Thema veröffentlicht und anschließend über den Webserver an Clients weitergeleitet wird, die über WebSockets verbunden sind.

In dieser Anleitung arbeiten Sie mit dem öffentlich verfügbaren Pub/Sub-Thema, das im Google Dataflow-Codelab "NYC Taxi Tycoon" verwendet wird. Dieses Thema beinhaltet einen Echtzeitstrom simulierter Taxi-Telemetriedaten, die auf Fahrtenaufzeichnungen aus den Datasets der Taxi & Limousine Commission aus New York City beruhen.

Architektur

Das nachstehende Diagramm veranschaulicht den Aufbau der Architektur, die Sie in dieser Anleitung erstellen.

Das Diagramm zeigt einen Nachrichten-Publisher außerhalb des Projekts, das die Compute Engine-Ressource enthält. Der Publisher sendet Nachrichten an ein Pub/Sub-Thema. Die Compute Engine-Instanz stellt die Nachrichten über WebSockets einem Browser zur Verfügung, in dem ein HTML5- und JavaScript-gestütztes Dashboard ausgeführt wird.

In dieser Anleitung werden verschiedene Tools kombiniert, um Pub/Sub und WebSockets zu verbinden:

• pulltop ist ein Node.js-Programm, das Sie im Verlauf dieser Anleitung installieren. Das Tool abonniert ein Pub/Sub-Thema und streamt die empfangenen Nachrichten an die Standardausgabe.
• websocketd ist ein kleines Befehlszeilentool, das eine vorhandene Befehlszeile einschließt und den Zugriff über ein WebSocket ermöglicht.

Durch die Kombination von pulltop und websocketd können Sie Nachrichten von dem Pub/Sub-Thema empfangen und über WebSockets an einen Browser streamen.

Durchsatz des Pub/Sub-Themas festlegen

Das öffentliche Pub/Sub-Thema von NYC Taxi Tycoon generiert 2.000 bis 2.500 simulierte Aktualisierungen von Taxifahrten pro Sekunde, was 8 MB oder mehr pro Sekunde entspricht. Die in Pub/Sub integrierte Ablaufsteuerung verringert den Nachrichtendurchsatz eines Abonnenten automatisch, wenn Pub/Sub eine wachsende Warteschlange nicht bestätigter Nachrichten erkennt. Daher kann der Nachrichtendurchsatz je nach Workstation, Netzwerkverbindung und Verarbeitungscode des Front-Ends stark variieren.

Effiziente Verarbeitung von Browsermeldungen

Angesichts des hohen Nachrichtenvolumens, das über den WebSocket-Stream eingeht, müssen Sie beim Schreiben des Frontend-Codes, der diesen Stream verarbeitet, sorgfältig vorgehen. Beispielsweise könnten Sie dynamisch HTML-Elemente für jede Nachricht erstellen. Bei dem erwarteten Nachrichtendurchsatz kann das Aktualisieren der Seite für jede Nachricht jedoch das Browserfenster blockieren. Häufige Arbeitsspeicherzuweisungen, die sich aus der dynamischen Erstellung von HTML-Elementen ergeben, verlängern auch die automatische Speicherbereinigung und beeinträchtigen den Nutzerkomfort. Mit anderen Worten, Sie werden nicht für jede der rund 2.000 Nachrichten, die pro Sekunde eingehen, document.createElement() aufrufen wollen.

In dieser Anleitung wird daher für die Verwaltung dieses dichten Nachrichtenstreams folgender Ansatz verwendet:

• Berechnung und kontinuierliche Aktualisierung einiger Streaming-Messwerte in Echtzeit. Dabei werden die meisten Informationen zu den beobachteten Nachrichten als zusammengefasste Werte angezeigt.
• Verwendung eines browsergestützten Dashboards zur Anzeige eines kleinen Auszugs der einzelnen Nachrichten nach einem vorprogrammierten Zeitplan, wobei nur Ein- und Ausstiegsereignisse in Echtzeit angezeigt werden.

Die nachstehende Abbildung zeigt das Dashboard, das im Rahmen dieser Anleitung erstellt wird.

Die Abbildung zeigt eine Latenz der letzten Nachricht von 24 Millisekunden bei einem Durchsatz von fast 2.100 Nachrichten pro Sekunde. Wenn die für die Verarbeitung jeder einzelnen Nachricht kritischen Codepfade nicht rechtzeitig abgeschlossen werden, nimmt die Anzahl der pro Sekunde beobachteten Nachrichten mit zunehmender Latenz der letzten Nachricht ab. Die Erfassung der Fahrt erfolgt mit der JavaScript API setInterval, die alle drei Sekunden ausgeführt wird. Damit wird verhindert, dass das Frontend während seiner Laufzeit eine zu hohe Anzahl von DOM-Elementen erstellt. Bei Durchsatzraten von mehr als 10 pro Sekunde kann der überwiegende Teil dieser Elemente in jedem Fall nicht mehr beobachtet werden.

Das Dashboard beginnt mit der Verarbeitung von Ereignissen in der Mitte des Streams, sodass bereits laufende Fahrten vom Dashboard als neu erkannt werden, sofern sie nicht zuvor erfasst wurden. Der Code verwendet ein assoziatives Array zum Speichern jeder beobachteten Fahrt, das durch den Wert ride_id indexiert wird, und entfernt den Bezug zur spezifischen Fahrt, wenn der Fahrgast ausgestiegen ist. Fahrten im Status "enroute" oder "pickup" fügen einen Bezug zum Array hinzu, es sei denn (im Fall von "enroute"), die Fahrt wurde bereits vorher beobachtet.

WebSocket-Server installieren und konfigurieren

Erstellen Sie zuerst die Compute Engine-Instanz, die Sie als WebSocket-Server nutzen werden. Installieren Sie auf dieser Instanz dann die Tools, die Sie später benötigen.

1. Legen Sie in Cloud Shell die Standardzone für Compute Engine fest. Im folgenden Beispiel wird us-central1-a verwendet, Sie können jedoch eine beliebige Zone einstellen.

   gcloud config set compute/zone us-central1-a
   

2. Erstellen Sie eine Compute Engine-Instanz mit dem Namen websocket-server in der Standardzone:

   gcloud compute instances create websocket-server --tags wss
   

3. Fügen Sie eine Firewall-Regel hinzu, die TCP-Traffic auf Port 8000 zu jeder als wss gekennzeichneten Instanz zulässt:

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

4. Stellen Sie bei Verwendung eines vorhandenen Projekts sicher, dass der TCP-Port 22 geöffnet ist, um die SSH-Verbindung zur Instanz zuzulassen.

   Standardmäßig ist die Firewallregel default-allow-ssh im Standardnetzwerk aktiviert. Wenn Sie oder Ihr Administrator die Standardregel in einem vorhandenen Projekt jedoch entfernt haben, ist der TCP-Port 22 möglicherweise nicht geöffnet. (Wenn Sie ein neues Projekt für diese Anleitung erstellt haben, ist die Regel standardmäßig aktiviert und Sie müssen nichts tun.)

   Fügen Sie eine Firewall-Regel hinzu, die TCP-Traffic auf Port 22 zu jeder als wss gekennzeichneten Instanz zulässt:

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

5. Stellen Sie über SSH eine Verbindung zur Instanz her:

   gcloud compute ssh websocket-server
   

6. Stellen Sie im Terminalbefehl der Instanz die Konten auf root, damit Sie die Software installieren können:

   sudo -s
   

7. Installieren Sie die Tools git und unzip.

   apt-get install -y unzip git
   

8. Installieren Sie die Binärdatei websocketd auf der Instanz:

   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
   

Installieren Sie Node.js und den Anleitungscode.

1. Installieren Sie in einem Terminal der Instanz Node.js:

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

2. Laden Sie das Quell-Repository der Anleitung herunter:

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

3. Ändern Sie die Berechtigungen in pulltop, um die Ausführung zuzulassen.

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

4. Installieren Sie die Abhängigkeiten für pulltop:

   cd pulltop
   npm install
   sudo npm link
   

Testen, ob pulltop Nachrichten lesen kann

1. Führen Sie auf der Instanz pulltop für das öffentliche Thema aus:

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

   Wenn pulltop funktioniert, wird ein Ergebnis-Stream wie der folgende angezeigt:

   {"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. Drücken Sie Ctrl+C, um den Stream zu stoppen.

Nachrichtenstream zu websocketd aufbauen

Nachdem Sie sich vergewissert haben, dass pulltop das Pub/Sub-Thema korrekt liest, können Sie den websocketd-Prozess starten, um Nachrichten an den Browser zu senden.

Nachrichten des Themas in einer lokalen Datei erfassen

In dieser Anleitung erfassen Sie den Nachrichtenstream von pulltop und schreiben ihn in eine lokale Datei. Für das Erfassen von Nachrichtenverkehr in einer lokalen Datei benötigen Sie zusätzlichen Speicher, andererseits wird dadurch der websocketd-Prozess vom Streaming der Pub/Sub-Nachrichten abgekoppelt. Die lokale Erfassung der Daten kann nützlich sein, wenn Sie das Pub/Sub-Streaming vorübergehend anhalten möchten (z. B. zum Einstellen von Parametern der Ablaufsteuerung), ohne ein Reset der aktuell verbundenen WebSocket-Clients erzwingen zu müssen. Nachdem der Nachrichtenstream wiederhergestellt ist, setzt websocketd das Streaming zu den Clients automatisch fort.

1. Führen Sie auf der Instanz pulltop für das öffentliche Thema aus und leiten Sie die Nachrichtenausgabe in die lokale Datei taxi.json weiter. Mit dem Befehl nohup wird das Betriebssystem angewiesen, den pulltop-Prozess weiter auszuführen, wenn Sie sich abmelden oder das Terminal schließen.

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

2. Prüfen Sie, ob in die Datei JSON-Nachrichten geschrieben werden:

   tail /var/tmp/taxi.json
   

   Wenn die Nachrichten in die Datei taxi.json geschrieben werden, sieht die Ausgabe etwa so aus:

   {"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. Wechseln Sie in den Webordner Ihrer Anwendung:

   cd ../web
   

4. Starten Sie websocketd, um den Inhalt der lokalen Datei mit WebSockets zu streamen:

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

   Damit wird der Befehl websocketd im Hintergrund ausgeführt. Das Tool websocketd erfasst die Ausgabe des Befehls tail und streamt die einzelnen Elemente als WebSocket-Nachrichten.

5. Prüfen Sie den Inhalt von nohup.out, um zu ermitteln, ob der Server korrekt gestartet wurde:

   tail nohup.out
   

   Wenn alles korrekt funktioniert, sieht die Ausgabe etwa so aus:

   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 : .
   

Nachrichten visualisieren

Nachrichten mit Daten individueller Fahrten, die im Pub/Sub-Thema veröffentlicht werden, haben eine Struktur wie diese:

{
  "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
}

Anhand dieser Werte können Sie verschiedene Messwerte für den Dashboard-Header berechnen. Die Berechnungen werden bei jedem eingehenden Fahrtereignis vorgenommen. Die Werte sind folgende:

• Latenz der letzten Nachricht: Anzahl der Sekunden, die zwischen dem Zeitstempel des letzten beobachteten Fahrtereignisses und der aktuellen Zeit verstrichen sind (erfasst mit der Uhr des Systems, auf dem der Webbrowser gehostet wird).
• Aktive Fahrten: Anzahl der aktuell laufenden Fahrten. Diese Zahl kann schnell steigen und sie nimmt ab, wenn der ride_status-Wert dropoff erfasst wird.
• Nachrichtendurchsatz: Durchschnittliche Anzahl der pro Sekunde verarbeiteten Fahrtereignisse.
• Gesamtstrecke: Summe der bei allen Fahrten zurückgelegten Meter. Diese Zahl nimmt ab, wenn Fahrten abgebrochen werden.
• Fahrgäste gesamt: Anzahl der Fahrgäste aller Fahrten. Diese Zahl nimmt zu, wenn Fahrten abgeschlossen werden.
• Durchschnittliche Fahrgastzahl pro Fahrt: Gesamtzahl der Fahrten, dividiert durch die Anzahl der Fahrgäste.
• Durchschnittliche Strecke pro Fahrgast: Gesamtstrecke, dividiert durch die Anzahl der Fahrgäste.

Zusätzlich zu den Messwerten und den Fahrtbeispielen wird im Dashboard über dem Raster der Fahrtbeispiele eine Meldung eingeblendet, wenn ein Fahrgast ein- oder aussteigt.

1. Rufen Sie die externe IP-Adresse der aktuellen Instanz ab:

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

2. Kopieren Sie die IP-Adresse.

3. Öffnen Sie auf Ihrem lokalen Computer einen Webbrowser und geben Sie diese URL ein:

   http://$ip-address:8000

   Sie sehen eine Seite mit dem Dashboard, das Sie mit dieser Anleitung erstellt haben:

   

4. Klicken Sie oben auf das Taxisymbol, um die Verbindung zum Stream herzustellen und mit der Verarbeitung der Nachrichten zu beginnen.

   Einzelne Fahrten werden als Auszug von jeweils neun aktiven Fahrten alle drei Sekunden angezeigt:

   

   Sie können jederzeit auf das Taxisymbol klicken, um den WebSocket-Stream zu starten oder zu stoppen. Wird die WebSocket-Verbindung getrennt, wird das Symbol rot und die Aktualisierung der einzelnen Fahrten und Messwerte wird unterbrochen. Klicken Sie noch einmal auf das Taxisymbol, um die Verbindung wiederherzustellen.

Leistung

Der nachstehende Screenshot zeigt den Leistungsmonitor der Chrome-Entwicklertools, während im Browser-Tab rund 2.100 Nachrichten pro Sekunde verarbeitet werden.

Die Nachrichtenweiterleitung erfolgt mit einer Latenz von rund 30 ms und die CPU-Auslastung beträgt rund 80 %. Die Speicherauslastung hat einen Mindestwert von 29 MB bei einer Gesamtzuweisung von 57 MB und schwankt frei.