Elementi costitutivi di applicazioni distribuite. Primo approccio

Nel passato Articolo Abbiamo esaminato i fondamenti teorici dell'architettura reattiva. È tempo di parlare di flussi di dati, di modi per implementare sistemi Erlang/Elixir reattivi e di modelli di messaggistica al loro interno:

• Richiedere risposta
• Risposta in blocchi di richiesta
• Risposta con richiesta
• Pubblica-iscriviti
• Pubblica-sottoscrivi invertito
• Distribuzione dei compiti

SOA, MSA e messaggistica

SOA, MSA sono architetture di sistema che definiscono le regole per costruire sistemi, mentre la messaggistica fornisce le primitive per la loro implementazione.

Non voglio promuovere questa o quella architettura di sistema. Sono favorevole all'utilizzo delle pratiche più efficaci e utili per un progetto e un'attività specifici. Qualunque sia il paradigma che scegliamo, è meglio creare blocchi di sistema con un occhio allo stile Unix: componenti con connettività minima, responsabili delle singole entità. I metodi API eseguono le azioni più semplici possibili con le entità.

La messaggistica è, come suggerisce il nome, un broker di messaggi. Il suo scopo principale è ricevere e inviare messaggi. È responsabile delle interfacce per l'invio delle informazioni, della formazione di canali logici per la trasmissione delle informazioni all'interno del sistema, dell'instradamento e del bilanciamento, nonché della gestione dei guasti a livello di sistema.
La messaggistica che stiamo sviluppando non cerca di competere o sostituire RabbitMQ. Le sue caratteristiche principali:

• Distribuzione.
  I punti di scambio possono essere creati su tutti i nodi del cluster, il più vicino possibile al codice che li utilizza.
• Semplicità.
  Concentrarsi sulla riduzione al minimo del codice standard e sulla facilità d'uso.
• Le migliori prestazioni.
  Non stiamo cercando di ripetere la funzionalità di RabbitMQ, ma evidenziamo solo il livello architetturale e di trasporto, che inseriamo nell'OTP nel modo più semplice possibile, minimizzando i costi.
• Flessibilità.
  Ogni servizio può combinare molti modelli di scambio.
• Resilienza fin dalla progettazione.
• Scalabilità.
  La messaggistica cresce con l'applicazione. All'aumentare del carico è possibile spostare i punti di scambio su singole macchine.

Nota. In termini di organizzazione del codice, i metaprogetti sono adatti per sistemi Erlang/Elixir complessi. Tutto il codice del progetto si trova in un unico repository: un progetto ombrello. Allo stesso tempo, i microservizi sono isolati al massimo ed eseguono operazioni semplici che sono responsabili di un'entità separata. Con questo approccio è facile mantenere l'API dell'intero sistema, è facile apportare modifiche, è conveniente scrivere test unitari e di integrazione.

I componenti del sistema interagiscono direttamente o tramite un broker. Dal punto di vista della messaggistica, ogni servizio ha diverse fasi di vita:

• Inizializzazione del servizio.
  In questa fase, il processo e le dipendenze che eseguono il servizio vengono configurati e avviati.
• Creazione di un punto di scambio.
  Il servizio può utilizzare un punto di scambio statico specificato nella configurazione del nodo, oppure creare punti di scambio in modo dinamico.
• Registrazione del servizio.
  Affinché il servizio possa soddisfare le richieste, è necessario che sia registrato presso il punto di scambio.
• Funzionamento normale.
  Il servizio produce lavoro utile.
• Completamento del lavoro.
  Sono possibili 2 tipi di spegnimento: normale ed emergenza. Durante il normale funzionamento il servizio viene disconnesso dal punto di scambio e si interrompe. In situazioni di emergenza, la messaggistica esegue uno degli script di failover.

Sembra piuttosto complicato, ma il codice non è poi così spaventoso. Esempi di codice con commenti verranno forniti poco dopo nell'analisi dei modelli.

Cambi Merce

Il punto di scambio è un processo di messaggistica che implementa la logica di interazione con i componenti all'interno del modello di messaggistica. In tutti gli esempi presentati di seguito, i componenti interagiscono attraverso punti di scambio, la cui combinazione forma la messaggistica.

Modelli di scambio di messaggi (MEP)

A livello globale, i modelli di scambio possono essere suddivisi in bidirezionali e unidirezionali. I primi implicano una risposta a un messaggio in arrivo, i secondi no. Un classico esempio di modello a due vie nell'architettura client-server è il modello richiesta-risposta. Diamo un'occhiata al modello e alle sue modifiche.

Richiesta-risposta o RPC

RPC viene utilizzato quando abbiamo bisogno di ricevere una risposta da un altro processo. Questo processo potrebbe essere in esecuzione sullo stesso nodo o trovarsi in un continente diverso. Di seguito è riportato un diagramma dell'interazione tra client e server tramite messaggistica.

Poiché la messaggistica è completamente asincrona, per il cliente lo scambio si divide in 2 fasi:

1. richiesta invio

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

   Exchange ‒ nome univoco del punto di scambio
   Tag di corrispondenza della risposta ‒ etichetta locale per l'elaborazione della risposta. Ad esempio, nel caso di invio di più richieste identiche appartenenti a utenti diversi.
   Definizione della richiesta - corpo richiesta
   HandlerProcess ‒ PID dell'handler. Questo processo riceverà una risposta dal server.

2. Elaborazione della risposta

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

   Carico utile della risposta - risposta del server.

Anche per il server il processo si compone di 2 fasi:

1. Inizializzazione del punto di scambio
2. Elaborazione delle richieste ricevute

Illustriamo questo modello con il codice. Diciamo che dobbiamo implementare un servizio semplice che fornisca un unico metodo per l'ora esatta.

Codice server

Definiamo l'API del servizio 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{}
}).

Definiamo il controller del servizio 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 inviare una richiesta al servizio, puoi chiamare l'API di richiesta di messaggistica ovunque nel client:

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

In un sistema distribuito, la configurazione dei componenti può essere molto diversa e al momento della richiesta, la messaggistica potrebbe non essere ancora avviata, oppure il controller del servizio non sarà pronto a soddisfare la richiesta. Pertanto, dobbiamo controllare la risposta del messaggio e gestire il caso di errore.
Dopo aver inviato con successo, il cliente riceverà una risposta o un errore dal servizio.
Gestiamo entrambi i 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};

Risposta in blocchi di richiesta

È meglio evitare di inviare messaggi enormi. Da questo dipende la reattività e il funzionamento stabile dell'intero sistema. Se la risposta a una query occupa molta memoria, è obbligatorio suddividerla in parti.

Permettetemi di farvi un paio di esempi di casi del genere:

• I componenti scambiano dati binari, come file. Suddividere la risposta in piccole parti ti aiuta a lavorare in modo efficiente con file di qualsiasi dimensione ed evitare overflow di memoria.
• Annunci. Ad esempio, dobbiamo selezionare tutti i record da un'enorme tabella nel database e trasferirli su un altro componente.

Chiamo queste risposte locomotiva. In ogni caso, 1024 messaggi da 1 MB sono meglio di un singolo messaggio da 1 GB.

Nel cluster Erlang otteniamo un ulteriore vantaggio: ridurre il carico sul punto di scambio e sulla rete, poiché le risposte vengono immediatamente inviate al destinatario, aggirando il punto di scambio.

Risposta con richiesta

Questa è una modifica piuttosto rara del modello RPC per la creazione di sistemi di dialogo.

Pubblica-sottoscrivi (albero di distribuzione dei dati)

I sistemi basati sugli eventi li consegnano ai consumatori non appena i dati sono pronti. Pertanto, i sistemi sono più inclini a un modello push che a un modello pull o poll. Questa funzionalità consente di evitare di sprecare risorse richiedendo e attendendo costantemente dati.
La figura mostra il processo di distribuzione di un messaggio ai consumatori iscritti a un argomento specifico.

Esempi classici dell'utilizzo di questo modello sono la distribuzione dello stato: il mondo di gioco nei giochi per computer, i dati di mercato sugli scambi, le informazioni utili nei feed di dati.

Diamo un'occhiata al codice abbonato:

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.

La fonte può chiamare la funzione per pubblicare un messaggio in qualsiasi posto conveniente:

messaging:publish_message(Exchange, Key, Message).

Exchange - nome del punto di scambio,
Le - chiave di instradamento
Messaggio - carico utile

Pubblica-sottoscrivi invertito

Espandendo pub-sub, puoi ottenere uno schema comodo per la registrazione. L'insieme di fonti e consumatori può essere completamente diverso. La figura mostra un caso con un consumatore e più fonti.

Modello di distribuzione dei compiti

Quasi tutti i progetti prevedono attività di elaborazione differite, come la generazione di report, la consegna di notifiche e il recupero di dati da sistemi di terze parti. Il throughput del sistema che esegue queste attività può essere facilmente scalato aggiungendo gestori. Tutto ciò che ci resta è formare un cluster di processori e distribuire uniformemente i compiti tra di loro.

Diamo un'occhiata alle situazioni che si presentano utilizzando l'esempio di 3 gestori. Anche nella fase di distribuzione dei compiti si pone la questione dell'equità della distribuzione e dell'eccesso di addetti. La distribuzione all'italiana sarà responsabile dell'equità e, per evitare una situazione di overflow degli handler, introdurremo una restrizione prefetch_limit. In condizioni transitorie prefetch_limit impedirà a un gestore di ricevere tutte le attività.

La messaggistica gestisce le code e la priorità di elaborazione. I processori ricevono le attività non appena arrivano. L'attività può essere completata con successo o fallire:

• messaging:ack(Tack) - chiamato se il messaggio viene elaborato correttamente
• messaging:nack(Tack) - chiamato in tutte le situazioni di emergenza. Una volta restituita l'attività, la messaggistica la passerà a un altro gestore.

Supponiamo che si sia verificato un errore complesso durante l'elaborazione di tre attività: il processore 1, dopo aver ricevuto l'attività, si è bloccato senza avere il tempo di segnalare nulla al punto di scambio. In questo caso, il punto di scambio trasferirà l'attività a un altro gestore una volta scaduto il timeout di ack. Per qualche motivo, l'handler 3 ha abbandonato l'attività e ha inviato nack; di conseguenza, l'attività è stata trasferita anche a un altro handler che l'ha completata con successo.

Risultato preliminare

Abbiamo trattato gli elementi costitutivi di base dei sistemi distribuiti e acquisito una conoscenza di base del loro utilizzo in Erlang/Elixir.

Combinando modelli di base, puoi costruire paradigmi complessi per risolvere problemi emergenti.

Nella parte finale della serie esamineremo le questioni generali relative all'organizzazione dei servizi, al routing e al bilanciamento e parleremo anche del lato pratico della scalabilità e della tolleranza ai guasti dei sistemi.

Fine della seconda parte.

foto Mario Christensen
Illustrazioni preparate utilizzando websequencediagrams.com

Fonte: habr.com