Stavebné bloky distribuovaných aplikácií. Prvý prístup

V poslednom článok Preskúmali sme teoretické základy reaktívnej architektúry. Je čas hovoriť o tokoch údajov, spôsoboch implementácie reaktívnych systémov Erlang/Elixir a vzoroch správ v nich:

• Žiadosť-odpoveď
• Request-Chunked Response
• Odpovedzte žiadosťou
• Publikovať-prihlásiť sa
• Obrátené Publikovať-prihlásiť sa
• Rozdelenie úloh

SOA, MSA a správy

SOA, MSA sú systémové architektúry, ktoré definujú pravidlá pre budovanie systémov, zatiaľ čo zasielanie správ poskytuje primitívy na ich implementáciu.

Nechcem propagovať tú či onú systémovú architektúru. Som za používanie najefektívnejších a najužitočnejších praktík pre konkrétny projekt a podnikanie. Bez ohľadu na to, akú paradigmu zvolíme, je lepšie vytvárať systémové bloky s ohľadom na Unixovú cestu: komponenty s minimálnou konektivitou, zodpovedné za jednotlivé entity. Metódy API vykonávajú najjednoduchšie možné akcie s entitami.

Messaging je, ako už názov napovedá, sprostredkovateľ správ. Jeho hlavným účelom je prijímať a odosielať správy. Je zodpovedný za rozhrania na odosielanie informácií, vytváranie logických kanálov na prenos informácií v rámci systému, smerovanie a vyvažovanie, ako aj za riešenie porúch na úrovni systému.
Správy, ktoré vyvíjame, sa nesnažia konkurovať alebo nahradiť rabbitmq. Jeho hlavné vlastnosti:

• Distribúcia.
  Výmenné body môžu byť vytvorené na všetkých uzloch klastra, čo najbližšie ku kódu, ktorý ich používa.
• Jednoduchosť.
  Zamerajte sa na minimalizáciu štandardného kódu a jednoduchosť použitia.
• Lepší výkon.
  Nesnažíme sa zopakovať funkčnosť králikamq, ale vyzdvihnúť iba architektonickú a transportnú vrstvu, ktorú čo najjednoduchšie vložíme do OTP, minimalizujeme náklady.
• Flexibilita.
  Každá služba môže kombinovať mnoho výmenných šablón.
• Odolnosť podľa návrhu.
• Škálovateľnosť.
  Správy rastú s aplikáciou. Pri zvyšovaní zaťaženia môžete presúvať výmenné body na jednotlivé stroje.

Poznámka. Z hľadiska organizácie kódu sú metaprojekty vhodné pre komplexné systémy Erlang/Elixir. Celý kód projektu sa nachádza v jednom úložisku – zastrešujúcom projekte. Mikroslužby sú zároveň maximálne izolované a vykonávajú jednoduché operácie, ktoré sú zodpovedné za samostatný subjekt. S týmto prístupom je ľahké udržiavať API celého systému, je ľahké robiť zmeny, je vhodné písať unit a integračné testy.

Systémové komponenty interagujú priamo alebo prostredníctvom makléra. Z pohľadu správ má každá služba niekoľko životných fáz:

• Inicializácia služby.
  V tejto fáze sa nakonfiguruje a spustí proces a závislosti vykonávajúce službu.
• Vytvorenie výmenného bodu.
  Služba môže používať statický výmenný bod špecifikovaný v konfigurácii uzla alebo dynamicky vytvárať výmenné body.
• Registrácia služby.
  Aby služba obsluhovala požiadavky, musí byť zaregistrovaná na výmennom mieste.
• Normálne fungovanie.
  Služba vytvára užitočnú prácu.
• Dokončenie práce.
  Možné sú 2 typy vypnutia: normálne a núdzové. Počas bežnej prevádzky sa služba odpojí od výmenného miesta a zastaví sa. V núdzových situáciách správ spustí jeden zo skriptov prepnutia pri zlyhaní.

Vyzerá to dosť komplikovane, ale kód nie je až taký strašidelný. Príklady kódu s komentármi budú uvedené v analýze šablón o niečo neskôr.

Výmeny

Výmenný bod je proces odosielania správ, ktorý implementuje logiku interakcie s komponentmi v rámci šablóny správ. Vo všetkých príkladoch uvedených nižšie komponenty interagujú prostredníctvom výmenných bodov, ktorých kombinácia vytvára správy.

Vzory výmeny správ (poslanci)

V globále možno vzory výmeny rozdeliť na obojsmerné a jednosmerné. Prvý znamená odpoveď na prichádzajúcu správu, druhý nie. Klasickým príkladom obojsmerného vzoru v architektúre klient-server je vzor požiadavka-odpoveď. Pozrime sa na šablónu a jej úpravy.

Request-Response alebo RPC

RPC sa používa, keď potrebujeme dostať odpoveď z iného procesu. Tento proces môže bežať na rovnakom uzle alebo sa môže nachádzať na inom kontinente. Nižšie je uvedený diagram interakcie medzi klientom a serverom prostredníctvom správ.

Keďže zasielanie správ je úplne asynchrónne, pre klienta je výmena rozdelená do 2 fáz:

1. odoslania požiadavky

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

   výmena ‒ jedinečný názov výmenného miesta
   ResponseMatchingTag ‒ lokálny štítok na spracovanie odpovede. Napríklad v prípade odoslania niekoľkých rovnakých požiadaviek patriacich rôznym používateľom.
   Požiadavka Definícia - telo žiadosti
   HandlerProcess ‒ PID handlera. Tento proces dostane odpoveď zo servera.

2. Spracovanie odpovede

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

   ResponsePayload - odpoveď servera.

V prípade servera proces pozostáva z 2 fáz:

1. Inicializácia výmenného bodu
2. Spracovanie prijatých žiadostí

Ilustrujme túto šablónu kódom. Povedzme, že potrebujeme implementovať jednoduchú službu, ktorá poskytuje jedinú metódu presného času.

Kód servera

Poďme definovať API služby v 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{}
}).

Definujme radič služby v 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.

Kód klienta

Ak chcete odoslať požiadavku do služby, môžete zavolať API žiadosti o správy kdekoľvek v klientovi:

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

V distribuovanom systéme môže byť konfigurácia komponentov veľmi odlišná a v čase požiadavky sa odosielanie správ ešte nemusí spustiť alebo servisný kontrolér nebude pripravený obslúžiť požiadavku. Preto musíme skontrolovať odozvu správ a vyriešiť prípad zlyhania.
Po úspešnom odoslaní dostane klient odpoveď alebo chybu zo služby.
Poďme zvládnuť oba prípady v 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};

Request-Chunked Response

Najlepšie je vyhnúť sa odosielaniu veľkých správ. Od toho závisí odozva a stabilná prevádzka celého systému. Ak odpoveď na dotaz zaberá veľa pamäte, rozdelenie na časti je povinné.

Dovoľte mi uviesť niekoľko príkladov takýchto prípadov:

• Komponenty si vymieňajú binárne údaje, ako sú súbory. Rozdelenie odozvy na malé časti vám pomôže efektívne pracovať so súbormi akejkoľvek veľkosti a vyhnúť sa preplneniu pamäte.
• Výpisy. Potrebujeme napríklad vybrať všetky záznamy z obrovskej tabuľky v databáze a preniesť ich do iného komponentu.

Tieto odpovede nazývam lokomotíva. V každom prípade je 1024 správ s veľkosťou 1 MB lepších ako jedna správa s veľkosťou 1 GB.

V klastri Erlang získame ďalšiu výhodu – zníženie zaťaženia výmenného bodu a siete, pretože odpovede sú okamžite odosielané príjemcovi a obchádzajú výmenný bod.

Odpovedzte žiadosťou

Toto je pomerne zriedkavá modifikácia vzoru RPC pre vytváranie dialógových systémov.

Publish-subscribe (strom distribúcie údajov)

Systémy riadené udalosťami ich doručia spotrebiteľom hneď, ako sú dáta pripravené. Systémy sú teda náchylnejšie na model push ako na model ťahania alebo prieskumu. Táto funkcia vám umožňuje vyhnúť sa plytvaniu zdrojmi neustálym vyžiadaním údajov a čakaním na ne.
Obrázok ukazuje proces distribúcie správy spotrebiteľom prihláseným na odber konkrétnej témy.

Klasickými príkladmi použitia tohto vzoru sú rozloženie stavu: herný svet v počítačových hrách, trhové údaje o burzách, užitočné informácie v dátových kanáloch.

Pozrime sa na kód predplatiteľa:

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.

Zdroj môže zavolať funkciu na zverejnenie správy na akomkoľvek vhodnom mieste:

messaging:publish_message(Exchange, Key, Message).

výmena - názov výmenného miesta,
Kľúč - smerovací kľúč
správa - užitočné zaťaženie

Obrátené Publikovať-prihlásiť sa

Rozšírením pub-sub môžete získať vzor vhodný na protokolovanie. Súbor zdrojov a spotrebiteľov môže byť úplne odlišný. Obrázok ukazuje prípad s jedným spotrebiteľom a viacerými zdrojmi.

Vzor distribúcie úloh

Takmer každý projekt zahŕňa odložené úlohy spracovania, ako je generovanie správ, doručovanie upozornení a získavanie údajov zo systémov tretích strán. Priepustnosť systému vykonávajúceho tieto úlohy sa dá jednoducho škálovať pridaním manipulátorov. Zostáva nám už len sformovať zhluk procesorov a medzi ne rovnomerne rozdeliť úlohy.

Pozrime sa na vzniknuté situácie na príklade 3 psovodov. Dokonca aj vo fáze rozdeľovania úloh vyvstáva otázka spravodlivosti rozdeľovania a pretečenia spracovateľov. Obojstranná distribúcia bude zodpovedná za férovosť a aby sme sa vyhli situácii preplnenia psovodov, zavedieme obmedzenie prefetch_limit. V prechodných podmienkach prefetch_limit zabráni jednému psovodovi prijímať všetky úlohy.

Správy riadia fronty a prioritu spracovania. Procesory dostávajú úlohy tak, ako prichádzajú. Úloha môže byť dokončená úspešne alebo zlyhá:

• messaging:ack(Tack) - volá sa, ak je správa úspešne spracovaná
• messaging:nack(Tack) - volaný vo všetkých núdzových situáciách. Akonáhle je úloha vrátená, správa ju odovzdá inému spracovateľovi.

Predpokladajme, že pri spracovaní troch úloh došlo ku komplexnému zlyhaniu: procesor 1 po prijatí úlohy spadol bez toho, aby stihol niečo nahlásiť výmennému bodu. V tomto prípade výmenný bod prenesie úlohu na iného handlera po uplynutí časového limitu potvrdenia. Z nejakého dôvodu sa handler 3 vzdal úlohy a poslal nack, v dôsledku čoho bola úloha prevedená aj na iného handlera, ktorý ju úspešne dokončil.

Predbežné zhrnutie

Pokryli sme základné stavebné bloky distribuovaných systémov a získali základné vedomosti o ich použití v Erlang/Elixir.

Kombináciou základných vzorov môžete vytvoriť zložité paradigmy na riešenie vznikajúcich problémov.

V záverečnej časti seriálu sa pozrieme na všeobecné otázky organizácie služieb, smerovania a vyvažovania a povieme si aj o praktickej stránke škálovateľnosti a odolnosti systémov proti chybám.

Koniec druhej časti.

fotografie Marius Christensen
Ilustrácie pripravené pomocou websequencediagrams.com

Zdroj: hab.com