🥇Građevinski blokovi distribuiranih aplikacija. Prvo približavanje

U prošloj članku Obradili smo teoretske osnove reaktivne arhitekture. Došlo je vrijeme da razgovaramo o tokovima podataka, načinima implementacije reaktivnih Erlang/Elixir sustava i obrascima razmjene poruka u njima:

Zahtjev-odgovor
Zahtjev-Chunkski odgovor
Odgovor s zahtjevom
Objavi-podijeli
Inverted Publish-subscribe
Distribucija zadataka

SOA, MSA i razmjena poruka

SOA, MSA – sistemske arhitekture koje definiraju pravila za izgradnju sustava, dok messaging pruža primitive za njihovu implementaciju.

Ne želim promicati određenu arhitekturu izgradnje sustava. Zalažem se za primjenu maksimalno učinkovitih i korisnih praksi za konkretan projekt i poslovanje. Koju god paradigmu odabrali, najbolje je kreirati sistemske blokove s pogledom na Unix-way: komponente s minimalnom povezanosti, odgovorne za pojedinačne entitete. API metode obavljaju maksimalno jednostavne radnje s entitetima.

Messaging ‒ kao što naziv sugerira ‒ je posrednik poruka. Njegov primarni cilj je primati i slati poruke. On je zadužen za sučelja slanja informacija, formiranje logičkih kanala za prijenos informacija unutar sustava, usmjeravanje i ravnotežu opterećenja, kao i obradu grešaka na sistemskoj razini.
Razvijeni messaging ne pokušava konkurirati rabbitmq-u niti ga zamijeniti. Njegove osnovne karakteristike su:

Distribuiranost.
Točke razmjene mogu se stvarati na svim čvorovima klastera, što bliže kodu koji ih koristi.
Jednostavnost.
Usmjerenost na minimiziranje šablonskog koda i jednostavnost korištenja.
Bolja izvedba.
Ne pokušavamo ponoviti funkcionalnost rabbitmq-a, već ističemo samo arhitektonski i transportni sloj koji se jednostavno uklapa u OTP, minimizirajući troškove.
Fleksibilnost.
Svaka usluga može kombinirati više obrazaca razmjene.
Otpornost na greške, ugrađena u dizajn.
Skalabilnost.
Messaging raste zajedno s aplikacijom. Kako se opterećenje povećava, točke razmjene se mogu prebaciti na posebne strojeve.

Napomena. S aspekta organizacije koda, za složene sustave na Erlangu/Elixiru dobro su prikladni meta-projekti. Sav kod projekta nalazi se u jednom repozitoriju - krovnom projektu. U ovom slučaju, mikroservisi su maksimalno izolirani i izvršavaju jednostavne operacije odgovorne za pojedinačne entitete. Ovim pristupom lako se održava API cijelog sustava, jednostavno se unose promjene, a praktično je pisati jedinčne i integracijske testove.

Komponente sustava međusobno komuniciraju izravno ili putem posrednika. S aspekta messaginga, svaka usluga ima nekoliko faza života:

Inicijalizacija usluge.
U ovoj fazi dolazi do konfiguracije i pokretanja izvršne usluge procesa i ovisnosti.
Kreiranje točke razmjene.
Usluga može koristiti statičku točku razmjene definiranu u konfiguraciji čvora, ili može dinamički kreirati točke razmjene.
Registracija usluge.
Kako bi usluga mogla obrađivati zahtjeve, potrebno ju je registrirati na točki razmjene.
Normalno djelovanje.
Usluga obavlja korisni rad.
Završetak rada.
Moguća su 2 načina završetka rada: redovni i izvanredni. U redovnom slučaju usluga se isključuje s točke razmjene i zaustavlja. U izvanrednim slučajevima messaging provodi jedan od scenarija obrade grešaka.

Izgleda prilično složeno, ali u kodu nije sve tako strašno. Primjeri koda s komentarima bit će prikazani u analizi predložaka malo kasnije.

Exchanges

Točka razmjene - proces messaginga koji implementira logiku interakcije s komponentama unutar predloška razmjene poruka. U svim primjerima predstavljenim u nastavku, komponente komuniciraju putem točaka razmjene, čija kombinacija oblikuje messaging.

Šabloni razmjene poruka (MEPs)

Globalno, šabloni razmjene mogu se podijeliti na dvosmjerne i jednosmjerne. Prvi podrazumijevaju odgovor na primljenu poruku, dok drugi ne. Klasični primjer dvosmjernog šablona u klijent-server arhitekturi je šablon zahtjeva-odgovora. Razmotrimo šablon i njegove modifikacije.

Zahtjev-odgovor ili RPC

RPC se koristi kada trebamo dobiti odgovor od drugog procesa. Ovaj proces može biti pokrenut na istom čvoru ili se nalaziti na drugom kontinentu. Ispod je prikazana shema interakcije klijenta i poslužitelja putem messaginga.

Budući da je messaging potpuno asinkron, klijentska razmjena dijeli se na 2 faze:

Slanje zahtjeva
```
messaging:request(Exchange, ResponseMatchingTag, RequestDefinition, HandlerProcess).
```
Razmjena ‒ jedinstveno ime razmjene.
ResponseMatchingTag ‒ lokalna oznaka za obradu odgovora. Na primjer, u slučaju slanja više identičnih zahtjeva koji pripadaju različitim korisnicima.
RequestDefinition ‒ tijelo zahtjeva.
HandlerProcess ‒ PID obrtnika. Ovaj proces će primiti odgovor s poslužitelja.

Obrada odgovora

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

ResponsePayload ‒ odgovor poslužitelja.

Za poslužitelj proces se također sastoji od 2 faze:

Inicijalizacija točke razmjene
Obrada pristiglih zahtjeva

Ilustrirat ćemo ovaj predložak kodom. Pretpostavimo da trebamo implementirati jednostavnu uslugu koja pruža jedinu metodu za točno vrijeme.

Kod poslužitelja

Izvadimo definiciju API-ja usluge u api.hrl:

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

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

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

Definirajmo kontroler usluge u time_controller.erl

%% U primjeru je prikodan samo značajan kod. Umetanjem u predložak gen_server možete dobiti radnu uslugu.

%% inicijalizacija gen_server
init(Args) ->
  %% povezivanje na točku razmjene
  messaging:monitor_exchange(req_resp, ?EXCHANGE, default, self())
  {ok, #{}}.

%% obrada događaja gubitka veze s točkom razmjene. Ovaj događaj dolazi ako točka razmjene još nije pokrenuta.
handle_info(#exchange_die{exchange = ?EXCHANGE}, State) ->
  erlang:send(self(), monitor_exchange),
  {noreply, State};

%% obrada API-ja
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};

%% završetak rada gen_server
terminate(_Reason, _State) ->
  messaging:demonitor_exchange(req_resp, ?EXCHANGE, default, self()),
  ok.

Kod klijenta

Da biste poslali zahtjev usluzi, bilo gdje u klijentu možete pozvati messaging request API:

case messaging:request(?EXCHANGE, tag, #time_req{opts = #{}}, self()) of
    ok -> ok;
    _ -> %% logika ponavljanja ili neuspjeha
end

U distribuiranom sustavu konfiguracija komponenti može biti različita, a tijekom zahtjeva messaging možda još nije pokrenut ili kontroler usluge neće biti spreman opslužiti zahtjev. Stoga moramo provjeriti odgovor messaging i obraditi slučaj neuspjeha.
Nakon uspješnog slanja, klijentu će od usluge stići odgovor ili pogreška.
Obradit ćemo oba slučaja u 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};

Zahtjev-Chunkski odgovor

Bolje je izbjeći slanje velikih poruka. To utječe na odziv i stabilnost cijelog sustava. Ako odgovor na zahtjev zauzima puno memorije, segmentacija je obavezna.

Navest ću nekoliko primjera takvih slučajeva:

Komponente razmjenjuju binarne podatke, na primjer datoteke. Segmentacija odgovora na manje dijelove pomaže učinkovito raditi s datotekama bilo koje veličine i sprječava preopterećenje memorije.
Listi. Na primjer, trebamo odabrati sve unose iz velike tablice u bazi i prenijeti ih drugoj komponenti.

Nazivam takve odgovore vlakom. U svakom slučaju, 1024 poruke od 1 MB bolje su od jedne poruke od 1 GB.

U Erlang klasteru dobivamo dodatnu korist - smanjenje opterećenja na mjestu razmjene i mreži, jer se odgovori odmah šalju primatelju, zaobilazeći točku razmjene.

Odgovor s zahtjevom

To je prilično rijetka modifikacija RPC obrasca za izgradnju dijalogue sustava.

Sustavi temeljeni na događajima isporučuju podatke potrošačima kad su ti podaci spremni. Na taj način, sustavi su skloniji push modelu nego pull ili poll. Ova karakteristika omogućuje da se ne troše resursi na stalno upitavanje i čekanje podataka.
Na slici je prikazan proces distribucije poruka potrošačima koji su se prijavili na određenu temu.

Klasični primjeri korištenja ovog obrasca uključuju distribuciju stanja: virtualnog svijeta u računalnim igrama, tržišnih podataka na burzama, korisnih informacija u data feedovima.

Razmotrit ćemo kod pretplatnika:

init(_Args) ->
  %% prijavljujemo se na razmjenik, ključ = key
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {ok, #{}}.

handle_info(#exchange_die{exchange = ?SUBSCRIPTION}, State) ->
  %% ako razmjenik nije dostupan, pokušavamo ponovo povezati
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {noreply, State};

%% obrađujemo primljene poruke
handle_info(#'$msg'{exchange = ?SUBSCRIPTION, message = Msg}, State) ->
  ?debugVal(Msg),
  {noreply, State};

%% prilikom zaustavljanja potrošača - odjavljujemo se s razmjenika
terminate(_Reason, _State) ->
  messaging:unsubscribe(?SUBSCRIPTION, key, tag, self()),
  ok.

Izvor može pozvati funkciju objavljivanja poruke na bilo kojem prikladnom mjestu:

messaging:publish_message(Exchange, Key, Message).

Razmjena ‒ naziv točke razmjene,
Ključ ‒ ključ usmjeravanja,
Poruka, ‒ payload.

Implementacijom pub-sub modela može se dobiti obrazac pogodan za logiranje. Skup izvora i potrošača može biti potpuno različit. Na slici je prikazan slučaj s jednim potrošačem i više izvora.

Obrazac raspodjele zadataka.

U gotovo svakom projektu javljaju se zadaci odgođenog izvršavanja, poput generiranja izvještaja, slanja obavijesti, dobivanja podataka iz vanjskih sustava. Propusnost sustava koji izvršava te zadatke lako se skalira dodavanjem obrađivača. Sve što trebamo je formirati klaster obrađivača i ravnomjerno raspodijeliti zadatke između njih.

Razmotrimo pojave na primjeru 3 obrađivača. Već u fazi raspodjele zadataka javlja se pitanje pravednosti i preopterećenja obrađivača. Za pravednost će se pobrinuti round-robin raspodjela, a kako bismo izbjegli situacije preopterećenja obrađivača, uvest ćemo ograničenje prefetch_limit. U prijelaznim fazama prefetch_limit neće dopustiti jednom obrađivaču da preuzme sve zadatke.

Messaging upravlja redovima i prioritetima obrade. Obrađivači dobivaju zadatke kako dolaze. Izvršenje zadatka može završiti uspješno ili neuspješno:

messaging:ack(Tack) ‒ poziva se u slučaju uspješnog obavljanja poruke.
messaging:nack(Tack) ‒ poziva se u svim izvanrednim situacijama. Nakon vraćanja zadatka, messaging će ga proslijediti drugom obrađivaču.

Pretpostavimo da se tijekom obrade tri zadatka dogodila složena greška: obrađivač 1 se srušio nakon što je primio zadatak, ne uspjevši ništa prijaviti točki razmjene. U tom slučaju, točka razmjene, nakon isteka ack timeout-a, prenijet će zadatak drugom obrađivaču. Obrađivač 3 iz nekog je razloga odbio zadatak i poslao nack, na kraju je i taj zadatak prešao drugom obrađivaču koji ga je uspješno izvršio.

Preliminarni zaključak

Obradili smo osnovne sastavnice distribuiranih sustava i stekli osnovno razumijevanje njihove primjene u Erlangu/Elixiru.

Kombiniranjem osnovnih obrazaca, moguće je izgraditi složene paradigme za rješavanje pojavljenih zadataka.

U završnom dijelu ciklusa razmotrit ćemo opća pitanja organizacije usluga, usmjeravanja i balansiranja, kao i praktične aspekte skalabilnosti i otpornosti sustava.

Kraj drugog dijela.

Fotografija Marius Christensen
Ilustracije su pripremljene pomoću websequencediagrams.com

Izvor: habr.com