🥇Jaotatud rakenduste ehitusplokid. Esimene lähenemine

Eelmises osas artiklis oleme käsitlenud reaktiivse arhitektuuri teoreetilisi aluseid. On aeg rääkida andmevoogudest, reaktiivsete Erlang/Elixir süsteemide rakenduste teedest ja sõnumivahetuse mustritest neis:

Päring-vastus
Päring-Chunked vastus
Vastus koos päringuga
Avalda-tellimine
Pööratud avalda-tellimine
Ülesannete jaotamine

SOA, MSA ja sõnumivahetus

SOA, MSA on süsteemi arhitektuurid, mis määratlevad süsteemide loomise reeglid, samas kui sõnumivahetus pakub selle rakendamiseks primitiive.

Ma ei soovi propageerida ühtegi teatud süsteemide arhitektuuri. Olen selle poolt, et rakendada konkreetse projekti ja äri jaoks võimalikult tõhusaid ja kasulikku praktikat. Ükskõik, millise paradigma me valime, on süsteemide plokkide loomine parem Unix-viisi kohaselt: minimaalsete seotud komponentidega, mis vastutavad eraldi üksuste eest. API meetodid täidavad võimalikult lihtsaid toiminguid üksustega.

Messaging ‒ nagu nimigi viitab ‒ on sõnumivahetusplatvorm. Selle peamine eesmärk on sõnumite vastuvõtmine ja edastamine. See vastutab teabe edastamise liideste, loogiliste teabeülekande kanalite loomise, marsruutimise ja koormuse tasakaalustamise ning süsteemi tasandil rikke töötlemise eest.
Kavandatav messaging ei püüa konkureerida rabbitmq või asendada seda. Selle põhijooned on:

Jaotatus.
Vahetuspunktide loomine on võimalik klastrite kõigil sõlmedel, maksimaalselt lähedal koodile, mis neid kasutab.
Lihtsus.
Suunatus шаблонн koodi vähendamisele ja kasutusmugavusele.
Parim jõudlus.
Me ei püüa reprodutseerida rabbitmq funktsionaalsust, vaid eristame ainult arhitektuuri ja transportkihi, mida integreerime võimalikult lihtsalt OTP-sse, minimeerides kulusid.
Paindlikkus.
Iga teenus võib sisaldada mitmeid vahetusmalle.
Veakindlus, mis on sisse ehitatud disaini.
Skaalautuvus.
Messaging kasvab koos rakendusega. Koormuse suurenedes saab vahetuspunktid viia eraldi masinatesse.

Märkus. Koodiorganisatsiooni vaatenurgast sobivad keerukate süsteemide jaoks Erlangi/Elixiri meta-projektid hästi. Kogu projekti kood asub ühes hoidlas ‒ katuseprojekti all. Samal ajal on mikroteenused maksimaalselt isoleeritud ja teevad lihtsaid operatsioone, mis vastutavad eraldi entiteedi eest. Sel viisil on süsteemi API-d lihtne hallata, muudatusi hõlbus teha ning mugav on kirjutada üksus- ja integratsiooniteste.

Süsteemi komponendid suhtlevad otse või kaudu vahendaja. Messaging'u seisukohalt on igal teenusel mitu elutsükli faasi:

Teenuse initsialiseerimine.
Sellel etapil toimub teenuse täitmise protsessi ja sõltuvuste konfigureerimine ja käivitamine.
Vahetuspunkti loomine.
Teenusel võib olla staatiline vahetuspunkt, mis on määratud sõlme konfiguratsioonis, või luua vahetuspunktid dünaamiliselt.
Teenuse registreerimine.
Kuna teenus peab sooritama päringud, tuleb see registreerida vahetuspunktis.
Normaalne töö.
Teenuse teostab kasulikku tööd.
Töö lõpetamine.
On võimalikud kaks töö lõpetamise viisi: regulaarne ja hädaolukord. Regulaarse lõpetamise korral katkestatakse teenus vahetuspunktiga ja peatutakse. Hädaolukordade puhul käivitab messaging ühe rikke töötlemise stsenaariumi.

See näeb välja üsna keeruline, kuid koodis pole kõik nii hull. Koodinäited koos kommentaaridega esitatakse natuke hiljem mallide analüüsis.

Vahetused

Vahetuspunkt on messaging protsess, mis rakendab loogikat komponentide vahel suhtlemiseks vahetusskeemis. Kõik alltoodud näidised suhtlevad läbi vahetuspunktide, mille kombinatsioon moodustab messaging'i.

Message exchange patterns (MEPs)

Globaalsetel mallidel saab jagada kahepoolseteks ja ühepoolseteks. Esimeseid iseloomustab vastus saadud sõnumile, teised mitte. Kliendiserveri arhitektuuri klassikaline näide kahepoolsetest mallidest on Request-response mall. Vaatleme mallide ja nende modifikatsioonide näit.

Request–response ehk RPC

RPC-d kasutatakse, kui meil on vaja saada vastus teiselt protsessilt. See protsess võib käivituda samas sõlmes või asuda teisel kontinendil. Allpool on toodud kliendi ja serverile messaging kaudu.

Kuna messaging on täielikult asünkroonne, jaguneb kliendi vahetus kaheks faasiks:

Päringu saatmine
```
messaging:request(Exchange, ResponseMatchingTag, RequestDefinition, HandlerProcess).
```
Vahetus ‒ vahetuse unikaalne nimi
ResponseMatchingTag ‒ lokaalne silt vastuse töötlemiseks. Näiteks, kui saadetakse mitu sama päringut erinevatelt kasutajatelt.
RequestDefinition ‒ päringu keha
HandlerProcess ‒ töötleja PID. Sellele protsessile saadetakse serverilt vastus.

Vastuse töötlemine

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

ResponsePayload ‒ serveri vastus.

Serveri jaoks koosneb protsess samuti kahest faasist:

Vahetuspunkti initsialiseerimine
Saabuvate päringute töötlemine

Illustreerime seda mallikoodiga. Oletame, et peame rakendama lihtsat teenust, mis pakub täpset aega.

Serveri kood

Võtame teenuse API määratlemise 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{}
}).

Määrame teenuse kontrolleri time_controller.erl

%% Näites on esitatud ainult oluline kood. Lisades selle gen_server mallile, saab töötava teenuse.

%% gen_serveri algatamine
init(Args) ->
  %% ühendamine vahetuspunktiga
  messaging:monitor_exchange(req_resp, ?EXCHANGE, default, self())
  {ok, #{}}.

%% vahetuspunktiga ühenduse katkemise sündmuse töötlemine. See sündmus tuleb ka siis, kui vahetuspunkti ei ole veel alustatud.
handle_info(#exchange_die{exchange = ?EXCHANGE}, State) ->
  erlang:send(self(), monitor_exchange),
  {noreply, State};

%% API töötlemine
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_serveri lõpetamine
terminate(_Reason, _State) ->
  messaging:demonitor_exchange(req_resp, ?EXCHANGE, default, self()),
  ok.

Kliendi kood

Et saata päring teenusele, võib kliendi igas kohas kutsuda esile messaging request API:

juhtum messaging:request(?EXCHANGE, tag, #time_req{opts = #{}}, self()) of
    ok -> ok;
    _ -> %% kordus või ebaõnnestumise loogika
lõpp

Jaotatud süsteemis võivad komponentide konfiguratsioonid olla väga erinevad ning messaging'i päringu hetkeks võib see veel mitte käivituda, või teenuse kontroller ei pruugi päringut teenindamiseks valmis olla. Seetõttu on meil vajalik kontrollida messaging'i vastust ja töödelda ebaõnnestumise juhtumit.
Pärast edukat saatmist saadab teenus kliendile vastuse või vea.
Töötleme mõlemat juhtumit funktsioonis 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};

Päring-Chunked vastus

Parim on vältida suurte sõnumite edastamist. See mõjutab kogu süsteemi reageerimisvõimet ja stabiilsust. Kui päringu vastus võtab palju mälu, on jagamine osadeks kohustuslik.

Tuletan meelde paar sellise juhtumi näidet:

Komponendid vahetavad binaarandmeid, näiteks faile. Vastuse jagamine väikesteks osadeks aitab tõhusalt töötada igasuguste suurustega failidega ja vältida mäluhäireid.
Loendid. Näiteks peame valima kõik kirjed suurst andmebaasi tabelist ja edastama need teisele komponendile.

Ma nimetan selliseid vastuseid rongiks. Ükskõik kuidas, 1024 sõnumit, igaühe maht 1 MB, on parem kui üksainus sõnum, mille maht on 1 GB.

Erlang klastris saavutame täiendava eelise - koormuse vähenemise vahetuspunktis ja võrgus, kuna vastused suunatakse kohe saajale, mööda vahetuspunkti.

Vastus koos päringuga

See on üsna harv RPC mustri modifikatsioon, mida kasutatakse dialoogisüsteemide loomisel.

Sündmuste poolt suunatud süsteemid edastavad andmed tarbijatele, kui need on valmis. Seetõttu kalduvad süsteemid pigem push-mudelile kui pull- või poll-mudelile. See omadus võimaldab mitte raisata ressursse, küsides pidevalt ja oodates andmeid.
Joonisel on kujutatud sõnumi levitamise protsess tarbijatele, kes on registreeritud teatud teema jaoks.

Klassikalised näited selle mustri kasutamisest on oleku levitamine: mängumaailma videomängudes, turuandmed börsidel, kasulik info andmevoogudes.

Vaatame tellija koodi:

init(_Args) ->
  %% liitume vahetusse, võti = key
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {ok, #{}}.

handle_info(#exchange_die{exchange = ?SUBSCRIPTION}, State) ->
  %% kui vahetuspunkt ei ole saadaval, proovime uuesti ühendust luua
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {noreply, State};

%% töötleme saabunud sõnumeid
handle_info(#'$msg'{exchange = ?SUBSCRIPTION, message = Msg}, State) ->
  ?debugVal(Msg),
  {noreply, State};

%% tarbija peatamisel lahkume vahetuspunktist
terminate(_Reason, _State) ->
  messaging:unsubscribe(?SUBSCRIPTION, key, tag, self()),
  ok.

Allikas võib kutsuda sõnumi avaldamise funktsiooni mistahes sobivas kohas:

messaging:publish_message(Exchange, Key, Message).

Vahetus ‒ vahetuspunkti nimi,
Avaõigus ‒ suunamisvõti
Sõnum ‒ koormus

Pub-sub'i rakendamisega saab luua mustri, mis sobib logimiseks. Allikate ja tarbijate komplekt võib olla täiesti erinev. Joonisel on kujutatud juhtumit, kus on üks tarbija ja mitu allikat.

Ülesannete jaotamise mustr

Peaaegu igas projektis tekivad viivitustega töötlemise ülesanded, nagu aruannete koostamine, teadete saatmine ja andmete hankimine kolmandate osapoolte süsteemidest. Süsteemi läbilaskevõimet, mis neid ülesandeid täidab, on kerge skaleerida, lisades töötlejad. Kõik, mis me peame tegema, on luua töötlejate klaster ja jaotada ülesanded nende vahel ühtlaselt.

Vaatame esilekerkivaid olukordi 3 töötleja näitel. Veel enne ülesannete jaotamist tõuseb õiglasuse ja ülekoormuse küsimus. Õiglasuse eest vastutab ringjaotus, ja et vältida töötlejate ülekoormust, kehtestame piiri prefetch_limit. Üleminekurežiimides prefetch_limit ei lase ühel töötlejal kõiki ülesandeid saada.

Sõnumihaldus haldab järjekordi ja töötlemise prioriteeti. Töötlejad saavad ülesandeid vastavalt nende saabumisele. Ülesande täitmine võib lõppeda kas edukalt või läbikukkumisega:

messaging:ack(Tack) ‒ kutsutakse ülesande eduka töötlemise korral
messaging:nack(Tack) ‒ kutsutakse esile kõigis ebatavalistes olukordades. Pärast ülesande tagastamist edastab sõnumiteenus selle teisele töötlejale.

Oletame, et kolme ülesande töötlemise käigus tekkis keeruline rike: töötleja 1 kukkus ülesande saamisel kokku, ilma et oleks vahetusele midagi edastanud. Sellisel juhul edastab vahetus pärast ack timeout'i lõppu ülesande teisele töötlejale. Töötleja 3 loobus mingil põhjusel ülesandest ja saatis nack, seetõttu läks ülesanne ka teisele töötlejale, kes selle edukalt täitis.

Esialgne kokkuvõte

Oleme kaasanud ja põhjalikult arutanud jaotatud süsteemide põhielemente ja saanud nende rakendamise kohta algteadmised Erlangis/Elixir'is.

Kombineerides põhivorme, on võimalik luua keerulisi paradigmasid, et lahendada esilekerkivaid probleeme.

Tsükli lõpus vaatame üldisi teenuste korraldamise, marsruutimise ja tasakaalustamise küsimusi ning arutame ka praktilisi külgi süsteemide skaleeritavuse ja talitlushäirete vastupidavuse osas.

Tsükli teine osa on lõppenud.

Foto Marius Christensen
Illustratsioonid on valmistatud websequencediagrams.com abil.

Allikas: habr.com