העברת הודעות Pub/Sub בסטרימינג באמצעות WebSockets

במדריך הזה מוצג אופן הטיפול בכמויות גדולות של נתונים נכנסים באפליקציית frontend – במקרה הזה, דף אינטרנט – כשמשתמשים ב-Google Cloud. במדריך הזה מתוארים כמה מהאתגרים של סטרימינג בנפח גבוה. במדריך הזה מופיעה אפליקציה לדוגמה שממחישה איך להשתמש ב-WebSockets כדי להציג נתונים של זרם צפוף של הודעות שפורסמו בנושא Pub/Sub, ולעבד אותן בזמן כדי לשמור על ביצועים טובים בחלק הקדמי של האתר.

המדריך הזה מיועד למפתחים שמכירים את התקשורת בין דפדפן לשרת באמצעות HTTP, ואת הכתיבה של אפליקציות frontend באמצעות HTML,‏ CSS ו-JavaScript. ההדרכה מתבססת על ההנחה שיש לכם ניסיון מסוים ב-Google Cloudושאתם מכירים את כלי שורת הפקודה של Linux.

מטרות

• יוצרים מכונה וירטואלית (VM) ומגדירים אותה עם הרכיבים הדרושים להזרמת נתוני התשלום של מינוי Pub/Sub ללקוחות בדפדפן.
• מגדירים תהליך במכונה הווירטואלית כדי להירשם לנושא Pub/Sub ולהוציא את ההודעות הנפרדות ליומן.
• מתקינים שרת אינטרנט כדי להציג תוכן סטטי ולהזרים פלט של פקודות Shell ללקוחות WebSocket.
• אפשר להציג את הצבירות של נתוני הסטרימינג של WebSocket ואת דוגמאות ההודעות הנפרדות בדפדפן באמצעות HTML,‏ CSS ו-JavaScript.

עלויות

במסמך הזה משתמשים ברכיבים הבאים של Google Cloud, והשימוש בהם כרוך בתשלום:

• Compute Engine
• Pub/Sub

כדי להעריך את ההוצאות בהתאם לתחזית השימוש שלכם, אתם יכולים להיעזר במחשבון העלויות.

משתמשים חדשים של Google Cloud ? יכול להיות שאתם זכאים לתקופת ניסיון בחינם.

לפני שמתחילים

נכנסים לחשבון Google Cloud . אם אתם משתמשים חדשים ב- Google Cloud, צרו חשבון כדי שתוכלו להעריך את הביצועים של המוצרים שלנו בתרחישים מהעולם האמיתי. לקוחות חדשים מקבלים בחינם גם קרדיט בשווי 300$ להרצה, לבדיקה ולפריסה של עומסי העבודה.

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

כדי להימנע מחיובים נוספים אחרי שסיימתם את המדריך, תוכלו למחוק את המשאבים שיצרתם. פרטים נוספים מופיעים במאמר בנושא הסרת המשאבים.

מבוא

ככל שיותר אפליקציות מאמצות מודלים מבוססי-אירועים, חשוב שאפליקציות frontend יוכלו ליצור חיבורים פשוטים עם מינימום חיכוך לשירותי ההודעות שמהווים את אבן הפינה של הארכיטקטורות האלה.

יש כמה אפשרויות להזרמת נתונים ללקוחות של דפדפני אינטרנט. האפשרות הנפוצה ביותר היא WebSockets. במדריך הזה מוסבר איך להתקין תהליך שרשום לזרם של הודעות שמתפרסמות בנושא Pub/Sub, ולנתב את ההודעות האלה דרך שרת האינטרנט בדרך ללקוחות שמחוברים באמצעות WebSockets.

במדריך הזה תעבדו עם נושא Pub/Sub שזמין לציבור ומשמש ב-NYC Taxi Tycoon Google Dataflow CodeLab. בנושא הזה מוצג זרם בזמן אמת של טלמטריה מסימולציה של מוניות, שמבוססת על נתוני נסיעות היסטוריים שנאספו בניו יורק ממערכי הנתונים של רשומות הנסיעות של Taxi & Limousine Commission (הוועדה למוניות ולימוזינות).

ארכיטקטורה

התרשים הבא מציג את הארכיטקטורה של המדריך שיוצרים במדריך הזה.

בתרשים מוצג מפרסם הודעות שנמצא מחוץ לפרויקט שמכיל את משאב Compute Engine. המפרסם שולח הודעות לנושא Pub/Sub. המכונה של Compute Engine מאפשרת להשתמש בהודעות דרך WebSockets בדפדפן שמריץ לוח בקרה שמבוסס על HTML5 ו-JavaScript.

במדריך הזה נעשה שימוש בשילוב של כלים כדי לגשר בין Pub/Sub לבין Websockets:

• ‫pulltop הוא תוכנית Node.js שמתקינים כחלק מהמדריך הזה. הכלי נרשם לנושא ב-Pub/Sub ומזרים את ההודעות שמתקבלות לפלט רגיל.
• ‫websocketd הוא כלי שורת פקודה קטן שעוטף תוכנית קיימת של ממשק שורת פקודה ומאפשר גישה אליה באמצעות WebSocket.

אם משלבים בין pulltop לבין websocketd, אפשר להזרים לדפדפן הודעות שמתקבלות מנושא Pub/Sub באמצעות WebSockets.

שינוי התפוקה של נושא Pub/Sub

נושא ה-Pub/Sub הציבורי NYC Taxi Tycoon יוצר 2,000 עד 2,500 עדכונים של נסיעות מוניות מסימולציה בשנייה – עד 8 MB או יותר בשנייה. הבקרה המובנית על זרימת הנתונים ב-Pub/Sub מאטה באופן אוטומטי את קצב ההודעות של מנוי אם המערכת מזהה תור גדל של הודעות שלא אושרו. לכן, יכול להיות שתראו שונות גבוהה בקצב העברת ההודעות בין תחנות עבודה שונות, חיבורי רשת וקוד עיבוד קצה.

עיבוד יעיל של הודעות בדפדפן

בגלל נפח ההודעות הגבוה שמגיע דרך זרם ה-WebSocket, צריך לחשוב היטב על כתיבת קוד ה-Frontend שמעבד את הזרם הזה. לדוגמה, אפשר ליצור באופן דינמי רכיבי HTML לכל הודעה. אבל בקצב ההודעות הצפוי, עדכון הדף לכל הודעה עלול לגרום לחלון הדפדפן להינעל. הקצאות תכופות של זיכרון שנובעות מיצירה דינמית של רכיבי HTML גם מאריכות את משך מנגנון איסוף הזבל, וכך פוגעות בחוויית המשתמש. בקיצור, לא כדאי להתקשר אל document.createElement() עבור כל אחת מ-2,000 ההודעות שמגיעות בכל שנייה.

הגישה שבה נעשה שימוש במדריך הזה לניהול זרם ההודעות הצפוף הזה היא כדלקמן:

• לחשב ולעדכן באופן שוטף קבוצה של מדדים של הזרם בזמן אמת, ולהציג את רוב המידע על ההודעות שנצפו כערכים מצטברים.
• אפשר להשתמש במרכז בקרה מבוסס-דפדפן כדי להמחיש מדגם קטן של הודעות נפרדות לפי לוח זמנים מוגדר מראש, ולהציג רק אירועים של מסירה ואיסוף בזמן אמת.

באיור הבא מוצג לוח הבקרה שנוצר כחלק מההדרכה הזו.

האיור מתאר זמן אחזור של ההודעה האחרונה של 24 אלפיות השנייה בקצב של כמעט 2,100 הודעות לשנייה. אם נתיבי הקוד הקריטיים לעיבוד כל הודעה בנפרד לא מסתיימים בזמן, מספר ההודעות שנצפו בשנייה יורד ככל שזמן האחזור של ההודעה האחרונה עולה. הדגימה של הנסיעות מתבצעת באמצעות קבוצת ה-API של JavaScript‏ setInterval שמוגדרת למחזור פעם בשלוש שניות, וכך נמנעת יצירה של מספר עצום של רכיבי DOM בחלק הקדמי של האתר במהלך חייו. (בכל מקרה, את רובם המכריע אי אפשר לראות בקצב גבוה מ-10 לשנייה).

לוח הבקרה מתחיל לעבד אירועים באמצע הזרם, ולכן נסיעות שכבר בעיצומן מזוהות כחדשות בלוח הבקרה, אלא אם הן כבר נראו קודם. הקוד משתמש במערך משויך כדי לאחסן כל נסיעה שנצפתה, עם אינדקס לפי הערך ride_id, ומסיר את ההפניה לנסיעה מסוימת כשהנוסע מגיע ליעד. נסיעות במצב 'בדרך' או 'איסוף' מוסיפות הפניה למערך הזה, אלא אם (במקרה של 'בדרך') הנסיעה נצפתה בעבר.

התקנה והגדרה של שרת WebSocket

כדי להתחיל, יוצרים מכונה של Compute Engine שתשמש כשרת WebSocket. אחרי שיוצרים את המופע, מתקינים בו את הכלים שצריך בהמשך.

1. ב-Cloud Shell, מגדירים את אזור ברירת המחדל של Compute Engine. בדוגמה הבאה מוצג us-central1-a, אבל אפשר להשתמש בכל אזור שרוצים.

   gcloud config set compute/zone us-central1-a
   

2. יוצרים מכונה של Compute Engine בשם websocket-server בתחום ברירת המחדל:

   gcloud compute instances create websocket-server --tags wss
   

3. מוסיפים כלל חומת אש שמאפשר תעבורת TCP ביציאה 8000 לכל מכונה שתויגה בתג wss:

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

4. אם אתם משתמשים בפרויקט קיים, ודאו שיציאת TCP‏ 22 פתוחה כדי לאפשר קישוריות SSH למופע.

   כברירת מחדל, כלל חומת האש default-allow-ssh מופעל ברשת שמוגדרת כברירת מחדל. עם זאת, אם אתם או האדמין שלכם הסרתם את כלל ברירת המחדל בפרויקט קיים, יכול להיות שהיציאה 22 של TCP לא פתוחה. (אם יצרתם פרויקט חדש לצורך המדריך הזה, הכלל מופעל כברירת מחדל ולא צריך לעשות כלום).

   מוסיפים כלל חומת אש שמאפשר תעבורת TCP ביציאה 22 לכל מכונה שתויגה בתור 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 מורה למערכת ההפעלה להמשיך להריץ את התהליך 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.
• Message rate (תעריף להודעה). המספר הממוצע של אירועי נסיעה שעברו עיבוד בכל שנייה.
• הסכום הכולל שחויב לפי שימוש. סכום המונים מכל הנסיעות הפעילות. המספר הזה יורד כשהנוסעים מגיעים ליעד.
• מספר הנוסעים הכולל. מספר הנוסעים בכל הנסיעות. המספר הזה יורד ככל שיותר נסיעות מושלמות.
• מספר הנוסעים הממוצע בכל נסיעה. המספר הכולל של הנסיעות, חלקי המספר הכולל של הנוסעים.
• הסכום הממוצע לתשלום לפי מונה לכל נוסע. הסכום הכולל של הנסיעה לפי מונה חלקי המספר הכולל של הנוסעים.

בנוסף למדדים ולדוגמאות של נסיעות בודדות, כשנוסע נאסף או מורד, מוצגת במרכז הבקרה התראה מעל רשת הדוגמאות של הנסיעות.

1. קבלת כתובת ה-IP החיצונית של המכונה הנוכחית:

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

2. מעתיקים את כתובת ה-IP.

3. במחשב המקומי, פותחים דפדפן אינטרנט חדש ומזינים את כתובת ה-URL:

   http://$ip-address:8000.

   יוצג דף עם לוח הבקרה של המדריך הזה:

   

4. לוחצים על סמל המונית בחלק העליון כדי לפתוח חיבור לסטרימינג ולהתחיל לעבד הודעות.

   הנסיעות מוצגות באופן ויזואלי עם מדגם של תשע נסיעות פעילות שמוצגות כל שלוש שניות:

   

   אפשר ללחוץ על סמל המונית בכל שלב כדי להתחיל או להפסיק את הזרמת הנתונים ב-WebSocket. אם חיבור ה-WebSocket נקטע, הסמל הופך לאדום והעדכונים של המדדים והנסיעות האישיות נעצרים. כדי להתחבר מחדש, לוחצים שוב על סמל המונית.

ביצועים

בצילום המסך הבא מוצג מדד הביצועים של הכלים למפתחים ב-Chrome בזמן שכרטיסיית הדפדפן מעבדת כ-2,100 הודעות בשנייה.

השליחה של ההודעות מתבצעת עם זמן אחזור של כ-30 אלפיות השנייה, והשימוש במעבד עומד בממוצע על כ-80%. השימוש בזיכרון מוצג בערך מינימלי של 29MB, עם הקצאה כוללת של 57MB, והוא גדל וקטן באופן חופשי.

הסרת המשאבים

הסרת כללים של חומת אש

אם השתמשתם בפרויקט קיים כדי לתרגל את המדריך הזה, תוכלו להסיר את כללי חומת האש שיצרתם. מומלץ לצמצם את מספר הפורטים הפתוחים.

1. מוחקים את כלל חומת האש שיצרתם כדי לאפשר TCP ביציאה 8000:

   gcloud compute firewall-rules delete websocket
   

2. אם יצרתם גם כלל חומת אש שמאפשר קישוריות SSH, מחקו את כלל חומת האש שמאפשר TCP ביציאה 22:

   gcloud compute firewall-rules delete wss-ssh
   

מחיקת הפרויקט

אם אתם לא רוצים להשתמש בפרויקט הזה יותר, אתם יכולים למחוק אותו.

זהירות: כשמוחקים פרויקט:
• התוכן של הפרויקט נמחק כולו. אם השתמשתם בפרויקט קיים כדי לתרגל את המשימות במדריך הזה, כשתמחקו אותו תימחק גם כל העבודה שביצעתם באותו פרויקט.
• מאבדים את מזהה הפרויקט בהתאמה אישית. כשיצרתם את הפרויקט, יכולתם לבחור לו מזהה פרויקט בהתאמה אישית כדי להשתמש בו בעתיד. כדי לשמור על כתובות ה-URL שמשתמשות במזהה הפרויקט, כמו appspot.com, צריך למחוק את המשאבים שנבחרו בפרויקט אבל לא את הפרויקט כולו.
1. במסוף Google Cloud , נכנסים לדף Manage resources.

   כניסה לדף Manage resources

2. ברשימת הפרויקטים, בוחרים את הפרויקט שרוצים למחוק ולוחצים על Delete.
3. כדי למחוק את הפרויקט, כותבים את מזהה הפרויקט בתיבת הדו-שיח ולוחצים על Shut down.

המאמרים הבאים

• מידע נוסף על Pub/Sub ועל פרוטוקול WebSockets
• כדי לאתר גיאוגרפית את הנסיעות שנאספו והסתיימו, צריך להוסיף את מפתח ה-API של Google Maps Platform אל cabdash.js.
• כדאי להעמיק את הקריאה ולהכיר דוגמאות לארכיטקטורות, תרשימים ושיטות מומלצות בנושאי Google Cloud. כל אלה זמינים במרכז הארכיטקטורה של Cloud.