Blocchi di costruzione di applicazioni distribuite. Prima avvicinamentu

In l'ultimu articulu Avemu esaminatu i fundamenti teorichi di l'architettura reattiva. Hè ora di parlà di flussi di dati, manere di implementà sistemi reattivi Erlang/Elixir è mudelli di messageria in elli:

• Richiesta-risposta
• Request-Chunked Response
• Risposta cù Richiesta
• Publish-subscribe
• Inverted Publish-subscribe
• Distribuzione di u travagliu

SOA, MSA è Messaging

SOA, MSA sò architetture di sistema chì definiscenu e regule per i sistemi di custruzzione, mentri a messageria furnisce primitivi per a so implementazione.

Ùn vogliu micca prumove questu o quellu architettura di sistema. Sò per aduprà e pratiche più efficaci è utili per un prughjettu specificu è affari. Qualunque paradigma scegliamu, hè megliu per creà blocchi di sistema cun un ochju nantu à u modu Unix: cumpunenti cù una cunnessione minima, rispunsevuli di entità individuali. I metudi API realizanu l'azzioni più simplici pussibuli cù entità.

A messageria hè, cum'è u nome suggerisce, un broker di messagi. U so scopu principale hè di riceve è mandà missaghji. Hè rispunsevuli di l'interfacce per l'invio di l'infurmazioni, a furmazione di canali lògichi per a trasmissione di l'infurmazioni in u sistema, u routing è l'equilibriu, è ancu a gestione di difetti à u livellu di u sistema.
A messageria chì sviluppemu ùn hè micca pruvatu à cumpete o rimpiazzà rabbitmq. E so caratteristiche principali:

• Distribuzione.
  I punti di scambiu ponu esse creati in tutti i nodi di cluster, u più vicinu pussibule à u codice chì l'utiliza.
• Simplicità.
  Focus nantu à minimizzà u codice boilerplate è facilità d'usu.
• Migliore prestazione.
  Ùn avemu micca pruvatu à ripetiri a funziunalità di rabbitmq, ma mette in risaltu solu a strata di l'architettura è di u trasportu, chì si mette in l'OTP u più simplice pussibule, minimizendu i costi.
• Flessibilità.
  Ogni serviziu pò cumminà parechji mudelli di scambiu.
• Resilienza da u disignu.
• Scalabilità.
  A messageria cresce cù l'applicazione. Quandu a carica aumenta, pudete spustà i punti di scambiu à e macchine individuali.

Rimarche. In quantu à l'urganizazione di u codice, i meta-prughjetti sò adattati per i sistemi Erlang / Elixir cumplessi. Tuttu u codice di u prugettu hè situatu in un repository - un prughjettu ombrello. À u listessu tempu, i microservizi sò massimu isolati è facenu operazioni simplici chì sò rispunsevuli di una entità separata. Cù questu approcciu, hè faciule di mantene l'API di tuttu u sistema, hè faciule fà cambiamenti, hè cunvenutu per scrive teste di unità è integrazione.

I cumpunenti di u sistema interagiscenu direttamente o attraversu un broker. Da una perspettiva di messageria, ogni serviziu hà parechje fasi di vita:

• Inizializazione di serviziu.
  In questu stadiu, u prucessu di esecutà u serviziu è e so dependenzii sò cunfigurati è lanciati.
• Crià un puntu di scambiu.
  U serviziu pò aduprà un puntu di scambiu staticu specificatu in a cunfigurazione di u nodu, o creà punti di scambiu dinamicamente.
• Registrazione di serviziu.
  Per u serviziu per serve e dumande, deve esse registratu à u puntu di scambiu.
• Funzionamentu normale.
  U serviziu pruduce un travagliu utile.
• Chjodi.
  Ci hè 2 tipi di chjusi pussibuli: normale è d'emergenza. Durante l'operazione normale, u serviziu hè disconnected from the exchange point and stops. In situazioni d'emergenza, a messageria eseguisce unu di i script di failover.

Sembra abbastanza cumplicatu, ma u codice ùn hè micca cusì spaventoso. Esempi di codice cù cumenti seranu datu in l'analisi di mudelli un pocu dopu.

barattu

U puntu di scambiu hè un prucessu di messageria chì implementa a logica di interazzione cù cumpunenti in u mudellu di messageria. In tutti l'esempii presentati quì sottu, i cumpunenti interagiscenu per mezu di punti di scambiu, a cumminazione di quale forma messageria.

Modelli di scambiu di messagi (MEPs)

In u mondu sanu, i mudelli di scambiu ponu esse divisi in bidirezionale è unidirezionale. I primi implicanu una risposta à un missaghju entrata, l'ultimi micca. Un esempiu classicu di un mudellu bidirezionale in l'architettura cliente-servitore hè u mudellu Request-response. Fighjemu u mudellu è e so mudificazioni.

Richiesta-risposta o RPC

RPC hè utilizatu quandu avemu bisognu di riceve una risposta da un altru prucessu. Stu prucessu pò esse in esecuzione nantu à u stessu node o situatu in un cuntinente diversu. Quì sottu hè un diagramma di l'interazzione trà u cliente è u servitore via messageria.

Siccomu a messageria hè cumplettamente asincrona, per u cliente u scambiu hè divisu in 2 fasi:

1. Mandà dumanda

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

   scanciu ‒ nome unicu di u puntu di scambiu
   RispostaMatchingTag ‒ etichetta locale per processà a risposta. Per esempiu, in u casu d'invià parechje dumande identiche chì appartenenu à diversi utilizatori.
   Definizione di dumanda - dumanda corpu
   Prucessu Handler ‒ PID di u gestore. Stu prucessu riceverà una risposta da u servitore.

2. Trattamentu di a risposta

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

   RispostaPayload - risposta di u servitore.

Per u servitore, u prucessu hè ancu custituitu da 2 fasi:

1. Initializing u puntu di scambiu
2. Trattamentu di e dumande ricevute

Illustremu stu mudellu cù codice. Diciamu chì avemu bisognu di implementà un serviziu simplice chì furnisce un solu metudu di tempu esatta.

U codice di u servitore

Definimu l'API di serviziu in 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{}
}).

Definimu u cuntrollu di serviziu in 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.

codice cliente

Per mandà una dumanda à u serviziu, pudete chjamà l'API di dumanda di messageria in ogni locu in u cliente:

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

In un sistema distribuitu, a cunfigurazione di cumpunenti pò esse assai sfarente è à u mumentu di a dumanda, a messageria ùn pò ancu principià, o u cuntrollu di u serviziu ùn serà micca prontu à serve a dumanda. Dunque, avemu bisognu di verificà a risposta di messageria è trattà u casu di fallimentu.
Dopu u mandatu successu, u cliente riceverà una risposta o errore da u serviziu.
Trattemu i dui casi in 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

Hè megliu per evità di mandà missaghji enormi. A rispunsibilità è u funziunamentu stabile di tuttu u sistema dipende da questu. Se a risposta à una dumanda occupa assai memoria, allora u split in parti hè ubligatoriu.

Lasciami dà un paru di esempi di tali casi:

• I cumpunenti scambià dati binari, cum'è i schedari. Rompe a risposta in parti chjuche vi aiuta à travaglià in modu efficiente cù i fugliali di qualsiasi dimensione è evità l'overflow di memoria.
• Liste. Per esempiu, avemu bisognu di selezziunà tutti i registri da una tavula enormosa in a basa di dati è trasfiriri à un altru cumpunente.

Chjamu sti risposti locomotiva. In ogni casu, 1024 missaghji di 1 MB sò megliu cà un missaghju unicu di 1 GB.

In u cluster Erlang, avemu un benefiziu supplementu - riducendu a carica nantu à u puntu di scambiu è a reta, postu chì e risposti sò immediatamente mandati à u destinatariu, sguassendu u puntu di scambiu.

Risposta cù Richiesta

Questa hè una mudificazione piuttostu rara di u mudellu RPC per custruisce sistemi di dialogu.

Publish-subscribe (arbre di distribuzione di dati)

I sistemi guidati da l'avvenimenti li consegnanu à i cunsumatori appena i dati sò pronti. Cusì, i sistemi sò più propensi à un mudellu push chè à un mudellu pull o poll. Sta funziunalità vi permette di evità di perdi risorse per sempre dumandendu è aspittendu dati.
A figura mostra u prucessu di distribuzione di un missaghju à i cunsumatori sottumessi à un tema specificu.

Esempi classici di usu di stu mudellu sò a distribuzione di u statu: u mondu di u ghjocu in i ghjoculi di computer, dati di u mercatu nantu à i scambii, infurmazioni utili in i feed di dati.

Fighjemu u codice di l'abbonatu:

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.

A fonte pò chjamà a funzione per publicà un missaghju in ogni locu cunvene:

messaging:publish_message(Exchange, Key, Message).

scanciu - nome di u puntu di scambiu,
Key - chjave di routing
missaghju - carica utile

Inverted Publish-subscribe

Expandendu pub-sub, pudete ottene un mudellu convenientu per u logu. U settore di fonti è cunsumatori pò esse cumplitamenti diffirenti. A figura mostra un casu cù un cunsumadore è parechje fonti.

U mudellu di distribuzione di u travagliu

Quasi ogni prughjettu implica compiti di trasfurmazioni differiti, cum'è a generazione di rapporti, a consegna di notificazioni è a ricuperazione di dati da sistemi di terzu. U throughput di u sistema chì eseguisce questi travaglii pò esse facilmente scalatu aghjunghjendu handlers. Tuttu ciò chì resta per noi hè di furmà un cluster di processori è distribuisce equitativamente i travaglii trà elli.

Fighjemu e situazioni chì si sviluppanu cù l'esempiu di 3 handlers. Ancu in u stadiu di a distribuzione di u travagliu, a quistione di l'equità di a distribuzione è u overflow di i manipulatori si pone. A distribuzione round-robin serà rispunsevuli di l'equità, è per evità una situazione di overflow di gestori, introduceremu una restrizione. prefetch_limit. In cundizioni transitori prefetch_limit impedisce à un gestore di riceve tutti i travaglii.

A messageria gestisce a fila è a priorità di trasfurmazioni. I prucessori ricevenu i travaglii quandu ghjunghjenu. U compitu pò esse cumpletu cù successu o fallu:

• messaging:ack(Tack) - chjamatu se u missaghju hè trattatu bè
• messaging:nack(Tack) - chjamatu in tutte e situazioni d'emergenza. Una volta u compitu hè tornatu, a messageria a trasmetterà à un altru gestore.

Suppone chì un fallimentu cumplessu hè accadutu durante u processu di trè compiti: u processatore 1, dopu avè ricivutu u compitu, s'hè lampatu senza avè u tempu di rapportà nunda à u puntu di scambiu. In questu casu, u puntu di scambiu trasfirerà u compitu à un altru gestore dopu chì u timeout ack hè scadutu. Per una certa ragione, u gestore 3 hà abbandunatu u compitu è ​​hà mandatu u nack; in u risultatu, u compitu hè statu ancu trasferitu à un altru gestore chì l'hà finitu cù successu.

Riassuntu preliminariu

Avemu cupertu i blocchi di custruzzione di basa di sistemi distribuiti è hà acquistatu una cunniscenza basica di u so usu in Erlang / Elixir.

Cumminendu mudelli basi, pudete custruisce paradigmi cumplessi per risolve i prublemi emergenti.

In l'ultima parte di a serie, avemu da fighjulà i prublemi generali di l'urganizazione di i servizii, u routing è l'equilibriu, è parlemu ancu di u latu praticu di a scalabilità è a toleranza di difetti di i sistemi.

Fine di a seconda parte.

Photo Marius Christensen
Illustrazioni preparate cù websequencediagrams.com

Source: www.habr.com