אבני בניין של אפליקציות מבוזרות. גישה ראשונה

בעבר статье בחנו את היסודות התיאורטיים של ארכיטקטורה ריאקטיבית. הגיע הזמן לדבר על זרימות נתונים, דרכים ליישם מערכות Erlang/Elixir תגובתיות ודפוסי הודעות בהן:

• בקשת תגובה
• בקשה-תגובה חתוכה
• תגובה עם בקשה
• פרסם-הירשם
• הפוך פרסם-הירשם
• חלוקת משימות

SOA, MSA והודעות

SOA, MSA הן ארכיטקטורות מערכת שמגדירות את הכללים לבניית מערכות, בעוד שהעברת הודעות מספקת פרימיטיביות ליישום שלהן.

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

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

• הפצה.
  ניתן ליצור נקודות החלפה בכל צמתי האשכול, קרוב ככל האפשר לקוד שמשתמש בהם.
• פַּשְׁטוּת.
  התמקד במזעור קוד ה-boilerplate ובנוחות השימוש.
• הופעה טובה יותר.
  אנחנו לא מנסים לחזור על הפונקציונליות של rabbitmq, אלא להדגיש רק את השכבה האדריכלית והתחבורתית, אותה אנו מתאימים ל-OTP בצורה פשוטה ככל האפשר, תוך מזעור עלויות.
• גְמִישׁוּת.
  כל שירות יכול לשלב תבניות החלפה רבות.
• חוסן בעיצוב.
• מדרגיות.
  העברת ההודעות גדלה עם האפליקציה. ככל שהעומס גדל, אתה יכול להעביר את נקודות ההחלפה למכונות בודדות.

REMARK. מבחינת ארגון הקוד, מטא-פרויקטים מתאימים היטב למערכות מורכבות של Erlang/Elixir. כל קוד הפרויקט ממוקם במאגר אחד - פרויקט גג. במקביל, שירותי מיקרו מבודדים בצורה מקסימלית ומבצעים פעולות פשוטות האחראיות על ישות נפרדת. בגישה זו קל לתחזק את ה-API של כל המערכת, קל לבצע שינויים, נוח לכתוב מבחני יחידות ואינטגרציה.

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

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

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

בורסות

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

דפוסי חילופי הודעות (חברי אירופה)

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

בקשה-תגובה או RPC

RPC משמש כאשר אנו צריכים לקבל תגובה מתהליך אחר. תהליך זה עשוי לפעול באותו צומת או ממוקם ביבשת אחרת. להלן תרשים של האינטראקציה בין הלקוח לשרת באמצעות הודעות.

מכיוון שהעברת ההודעות היא אסינכרונית לחלוטין, עבור הלקוח המרכזייה מחולקת ל-2 שלבים:

1. שולח בקשה

   messaging:request(Exchange, ResponseMatchingTag, RequestDefinition, HandlerProcess).

   המרת ‒ שם ייחודי של נקודת ההחלפה
   ResponseMatchingTag ‒ תווית מקומית לעיבוד התגובה. למשל, במקרה של שליחת מספר בקשות זהות השייכות למשתמשים שונים.
   RequestDefinition - גוף הבקשה
   HandlerProcess ‒ PID של המטפל. תהליך זה יקבל תגובה מהשרת.

2. מעבד את התגובה

   handle_info(#'$msg'{exchange = EXCHANGE, tag = ResponseMatchingTag,message = ResponsePayload}, State)

   ResponsePayload - תגובת שרת.

עבור השרת, התהליך מורכב גם מ-2 שלבים:

1. אתחול נקודת ההחלפה
2. טיפול בבקשות שהתקבלו

בואו נמחיש את התבנית הזו עם קוד. נניח שעלינו ליישם שירות פשוט המספק שיטת זמן אחת מדויקת.

קוד שרת

בואו נגדיר את ה-API של השירות ב-api.hrl:

%% =====================================================
%%  entities
%% =====================================================
-record(time, {
  unixtime :: non_neg_integer(),
  datetime :: binary()
}).

-record(time_error, {
  code :: non_neg_integer(),
  error :: term()
}).

%% =====================================================
%%  methods
%% =====================================================
-record(time_req, {
  opts :: term()
}).
-record(time_resp, {
  result :: #time{} | #time_error{}
}).

בואו נגדיר את בקר השירות ב-time_controller.erl

%% В примере показан только значимый код. Вставив его в шаблон gen_server можно получить рабочий сервис.

%% инициализация gen_server
init(Args) ->
  %% подключение к точке обмена
  messaging:monitor_exchange(req_resp, ?EXCHANGE, default, self())
  {ok, #{}}.

%% обработка события потери связи с точкой обмена. Это же событие приходит, если точка обмена еще не запустилась.
handle_info(#exchange_die{exchange = ?EXCHANGE}, State) ->
  erlang:send(self(), monitor_exchange),
  {noreply, State};

%% обработка API
handle_info(#time_req{opts = _Opts}, State) ->
  messaging:response_once(Client, #time_resp{
result = #time{ unixtime = time_utils:unixtime(now()), datetime = time_utils:iso8601_fmt(now())}
  });
  {noreply, State};

%% завершение работы gen_server
terminate(_Reason, _State) ->
  messaging:demonitor_exchange(req_resp, ?EXCHANGE, default, self()),
  ok.

קוד לקוח

על מנת לשלוח בקשה לשירות, ניתן להתקשר ל-API של בקשה להודעות בכל מקום בלקוח:

case messaging:request(?EXCHANGE, tag, #time_req{opts = #{}}, self()) of
    ok -> ok;
    _ -> %% repeat or fail logic
end

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

handle_info(#'$msg'{exchange = ?EXCHANGE, tag = tag, message = #time_resp{result = #time{unixtime = Utime}}}, State) ->
  ?debugVal(Utime),
  {noreply, State};

handle_info(#'$msg'{exchange = ?EXCHANGE, tag = tag, message = #time_resp{result = #time_error{code = ErrorCode}}}, State) ->
  ?debugVal({error, ErrorCode}),
  {noreply, State};

בקשה-תגובה חתוכה

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

הרשו לי לתת לכם כמה דוגמאות למקרים כאלה:

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

אני קורא לתגובות האלה קטר. בכל מקרה, 1024 הודעות של 1 MB עדיפות על הודעה בודדת של 1 GB.

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

תגובה עם בקשה

זהו שינוי נדיר למדי של תבנית ה-RPC לבניית מערכות דיאלוג.

פרסם-הירשם (עץ הפצת נתונים)

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

דוגמאות קלאסיות לשימוש בדפוס זה הן התפלגות המדינה: עולם המשחקים במשחקי מחשב, נתוני שוק בבורסות, מידע שימושי בהזנות נתונים.

בואו נסתכל על קוד המנוי:

init(_Args) ->
  %% подписываемся на обменник, ключ = key
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {ok, #{}}.

handle_info(#exchange_die{exchange = ?SUBSCRIPTION}, State) ->
  %% если точка обмена недоступна, то пытаемся переподключиться
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {noreply, State};

%% обрабатываем пришедшие сообщения
handle_info(#'$msg'{exchange = ?SUBSCRIPTION, message = Msg}, State) ->
  ?debugVal(Msg),
  {noreply, State};

%% при остановке потребителя - отключаемся от точки обмена
terminate(_Reason, _State) ->
  messaging:unsubscribe(?SUBSCRIPTION, key, tag, self()),
  ok.

המקור יכול לקרוא לפונקציה לפרסום הודעה בכל מקום נוח:

messaging:publish_message(Exchange, Key, Message).

המרת - שם נקודת ההחלפה,
מפתח - מפתח ניתוב
הערות נוספות - מטען

הפוך פרסם-הירשם

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

דפוס חלוקת משימות

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

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

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

• messaging:ack(Tack) - נקרא אם ההודעה עובדת בהצלחה
• messaging:nack(Tack) - נקרא בכל מצבי חירום. לאחר החזרת המשימה, העברת ההודעות תעביר אותה למטפל אחר.

נניח שכשל מורכב התרחש בזמן עיבוד שלוש משימות: מעבד 1, לאחר קבלת המשימה, קרס מבלי שהספיק לדווח על דבר לנקודת ההחלפה. במקרה זה, נקודת ההחלפה תעביר את המשימה למטפל אחר לאחר פקיעת הזמן הקצוב ל-ack. מסיבה כלשהי, מטפל 3 נטש את המשימה ושלח נאק; כתוצאה מכך, המשימה הועברה גם למטפל אחר שסיים אותה בהצלחה.

סיכום ראשוני

כיסינו את אבני הבניין הבסיסיות של מערכות מבוזרות והשגנו הבנה בסיסית של השימוש בהן ב-Erlang/Elixir.

על ידי שילוב של דפוסים בסיסיים, אתה יכול לבנות פרדיגמות מורכבות כדי לפתור בעיות מתעוררות.

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

סוף החלק השני.

תצלום מריוס כריסטנסן
איורים שהוכנו באמצעות websequencediagrams.com

מקור: www.habr.com