🥇Blocchi costitutivi delle applicazioni distribuite. Prima approssimazione

Nel passato articolo Abbiamo analizzato i fondamenti teorici dell'architettura reattiva. È tempo di parlare dei flussi di dati, dei modi di implementazione dei sistemi Erlang/Elixir reattivi e dei modelli di scambio di messaggi al loro interno:

Richiesta-risposta
Risposta a chunk di richiesta
Risposta con richiesta
Pubblica-sottoscrivi
Pubblica-sottoscrivi invertito
Distribuzione dei task

SOA, MSA e scambio di messaggi

SOA e MSA sono architetture sistemiche che definiscono le regole di costruzione dei sistemi, mentre il messaging fornisce i primitivi per la loro realizzazione.

Non voglio promuovere una particolare architettura di sistema. Sono a favore dell'applicazione delle pratiche più efficaci e utili per un progetto specifico e un'azienda. Qualunque paradigma scegliamo, è meglio creare i blocchi di sistema seguendo il principio Unix: componenti con bassa accoppiatura, responsabili di entità singole. I metodi API eseguono azioni il più semplici possibile con le entità.

Messaging ‒ come suggerisce il nome ‒ è un broker di messaggi. Il suo obiettivo principale è ricevere e inviare messaggi. Si occupa delle interfacce per l'invio delle informazioni, della formazione di canali logici di trasmissione all'interno del sistema, del routing e del bilanciamento, oltre a gestire i fallimenti a livello di sistema.
Il messaging sviluppato non cerca di competere con rabbitmq né di sostituirlo. Le sue principali caratteristiche sono:

Distribuzione.
I punti di interscambio possono essere creati su tutti i nodi del cluster, il più vicino possibile al codice che li utilizza.
Semplicità.
Orientato a minimizzare il codice boilerplate e a garantire la facilità d'uso.
Migliore performance.
Non cerchiamo di replicare la funzionalità di rabbitmq, ma ci concentriamo solo sul livello architetturale e di trasporto, che integriamo nel modo più semplice in OTP, minimizzando i costi.
Flessibilità.
Ogni servizio può combinare molti schemi di scambio.
Affidabilità, integrata nel design.
Scalabilità.
Il messaging cresce con l'applicazione. Con l'aumento del carico, è possibile spostare i punti di scambio su macchine separate.

Nota. Dal punto di vista dell'organizzazione del codice, i meta-progetti sono ideali per sistemi complessi su Erlang/Elixir. Tutto il codice del progetto si trova in un unico repository – un progetto ombrello. I microservizi, in questo caso, sono massimamente isolati e svolgono operazioni semplici, ciascuna responsabile di un'entità separata. Con questo approccio, è facile mantenere l'API dell'intero sistema, apportare modifiche e scrivere test unitari e di integrazione in modo conveniente.

I componenti del sistema interagiscono direttamente o attraverso un broker. Dal punto di vista del messaging, ciascun servizio ha diverse fasi della vita:

Inizializzazione del servizio.
In questa fase avviene la configurazione e l'avvio del processo di esecuzione del servizio e delle sue dipendenze.
Creazione di un punto di scambio.
Il servizio può utilizzare un punto di scambio statico, definito nella configurazione del nodo, o creare punti di scambio dinamicamente.
Registrazione del servizio.
Affinché un servizio possa gestire le richieste, deve essere registrato presso il punto di scambio.
Funzionamento normale.
Il servizio svolge un lavoro utile.
Fine del servizio.
Sono possibili due tipi di conclusione del lavoro: ordinaria e d'emergenza. Nella modalità ordinaria, il servizio si scollega dal punto di interscambio e si ferma. In caso di emergenze, il messaging esegue uno degli scenari di gestione degli errori.

Sembra piuttosto complicato, ma nel codice non è così spaventoso. Esempi di codice con commenti saranno forniti nell'analisi dei modelli più avanti.

Exchanges

Il punto di interscambio è un processo di messaging che implementa la logica di interazione con i componenti nell'ambito del modello di scambio di messaggi. In tutti gli esempi presentati di seguito, i componenti interagiscono attraverso punti di interscambio, la combinazione dei quali forma il messaging.

Message exchange patterns (MEPs)

Globalmente, i modelli di scambio possono essere suddivisi in bidirezionali e unidirezionali. I primi implicano una risposta al messaggio ricevuto, i secondi no. Un esempio classico di modello bidirezionale nell'architettura client-server è il modello Request-response. Esaminiamo il modello e le sue modifiche.

Request–response o RPC

RPC è utilizzato quando abbiamo bisogno di ricevere una risposta da un altro processo. Questo processo può essere eseguito sullo stesso nodo o trovarsi su un altro continente. Di seguito è riportato uno schema di interazione tra il client e server attraverso messaging.

Poiché messaging è completamente asincrono, il trasferimento per il client è diviso in 2 fasi:

Invio della richiesta
```
messaging:request(Exchange, ResponseMatchingTag, RequestDefinition, HandlerProcess).
```
Scambio ‒ nome unico del punto di scambio
ResponseMatchingTag ‒ etichetta locale per l'elaborazione della risposta. Ad esempio, nel caso di invio di più richieste identiche appartenenti a utenti diversi.
RequestDefinition ‒ corpo della richiesta
HandlerProcess ‒ PID del gestore. A questo processo arriverà la risposta dal server.

Elaborazione della risposta

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

ResponsePayload ‒ risposta dal server.

Per il server, il processo è anch'esso composto da 2 fasi:

Inizializzazione del punto di scambio
Elaborazione delle richieste ricevute

Illustriamo con il codice questo modello. Supponiamo che dobbiamo implementare un semplice servizio che fornisce un unico metodo per il tempo esatto.

Codice del server

Estraiamo la definizione dell'API del servizio in api.hrl:

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

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

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

Definiamo il controller del servizio in time_controller.erl

%% L'esempio mostra solo il codice significativo. Inserendolo nel template gen_server è possibile ottenere un servizio funzionante.

%% inizializzazione del gen_server
init(Args) ->
  %% collegamento al punto di scambio
  messaging:monitor_exchange(req_resp, ?EXCHANGE, default, self())
  {ok, #{}}.

%% gestione dell'evento di perdita di connessione con il punto di scambio. Questo stesso evento arriva se il punto di scambio non è ancora stato avviato.
handle_info(#exchange_die{exchange = ?EXCHANGE}, State) ->
  erlang:send(self(), monitor_exchange),
  {noreply, State};

%% gestione dell'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};

%% conclusione del lavoro del gen_server
terminate(_Reason, _State) ->
  messaging:demonitor_exchange(req_resp, ?EXCHANGE, default, self()),
  ok.

Codice cliente

Per inviare una richiesta al servizio, in qualsiasi parte del client è possibile chiamare l'API di richiesta messaging:

case messaging:request(?EXCHANGE, tag, #time_req{opts = #{}}, self()) of
    ok -> ok;
    _ -> %% logica di ripetizione o fallimento
end

In un sistema distribuito, la configurazione dei componenti può variare e al momento della richiesta messaging potrebbe non essere ancora avviata, oppure il controller del servizio non sarà pronto a gestire la richiesta. Pertanto, è necessario verificare la risposta del messaging e gestire il caso di errore.
Dopo l'invio riuscito, il cliente riceverà una risposta o un errore dal servizio.
Gestiremo 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 a chunk di richiesta

È meglio evitare di trasmettere messaggi di grandi dimensioni. Questo influisce sulla reattività e sulla stabilità dell'intero sistema. Se la risposta alla richiesta occupa molta memoria, è necessario frammentarla.

Ecco un paio di esempi di tali situazioni:

I componenti scambiano dati binari, come file. Frammentare la risposta in piccole parti aiuta a lavorare in modo efficace con file di qualsiasi dimensione e a evitare le sovraccariche di memoria.
Elenco. Ad esempio, dobbiamo selezionare tutte le voci da un enorme database e passarle a un altro componente.

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

Nel cluster Erlang otteniamo un vantaggio aggiuntivo: riduzione del carico sul punto di scambio e sulla rete, poiché le risposte vengono inviate direttamente al destinatario, bypassando il punto di scambio.

Risposta con richiesta

Questa è una modifica piuttosto rara del pattern RPC per costruire sistemi dialogici.

I sistemi orientati agli eventi consegnano i dati ai consumatori man mano che sono pronti. Pertanto, i sistemi sono più inclini a un modello push piuttosto che pull o poll. Questa caratteristica consente di non sprecare risorse attendendo costantemente i dati.
Nell'immagine è mostrato il processo di diffusione del messaggio ai consumatori che si sono iscritti a un determinato argomento.

Esempi classici dell'uso di questo pattern includono la diffusione dello stato: il mondo di gioco nei videogiochi, i dati di mercato nelle borse, informazioni utili nei data feed.

Esaminiamo il codice del sottoscrittore:

init(_Args) ->
  %% Iscriviti al messaggero, chiave = key
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {ok, #{}}.

handle_info(#exchange_die{exchange = ?SUBSCRIPTION}, State) ->
  %% se il punto di scambio non è disponibile, cerchiamo di riconnetterci
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {noreply, State};

%% elaboriamo i messaggi ricevuti
handle_info(#'$msg'{exchange = ?SUBSCRIPTION, message = Msg}, State) ->
  ?debugVal(Msg),
  {noreply, State};

%% alla fermata del consumatore - ci disconnettiamo dal punto di scambio
terminate(_Reason, _State) ->
  messaging:unsubscribe(?SUBSCRIPTION, key, tag, self()),
  ok.

La fonte può invocare la funzione di pubblicazione del messaggio in qualsiasi punto comodo:

messaging:publish_message(Exchange, Key, Message).

Scambio ‒ nome del punto di scambio,
Chiave ‒ chiave di instradamento
Messaggio ‒ payload

Implementando pub-sub, si può ottenere uno schema utile per la registrazione. L'insieme di fonti e consumatori può essere completamente diverso. L'immagine mostra un caso con un consumatore e molte fonti.

Schema di distribuzione dei task

In quasi ogni progetto si presentano compiti di elaborazione differita, come la generazione di report, l'invio di notifiche e il recupero di dati da sistemi esterni. La capacità del sistema che esegue questi compiti è facilmente scalabile aggiungendo gestori. Tutto ciò che dobbiamo fare è creare un cluster di gestori e distribuire equamente i compiti tra di loro.

Consideriamo le situazioni che possono sorgere con 3 gestori. Già nella fase di distribuzione dei compiti sorge la questione dell'equità della distribuzione e del sovraccarico dei gestori. L'equità sarà garantita dalla distribuzione round-robin, mentre per evitare che un gestore si sovraccarichi, introdurremo un limite prefetch_limit. Nei regimi transitori prefetch_limit non permetterà a un solo gestore di ricevere tutti i compiti.

Messaging gestisce le code e la priorità di elaborazione. I gestori ricevono le attività man mano che arrivano. L'esecuzione di un compito può concludersi con successo o con un rifiuto:

messaging:ack(Tack) ‒ viene chiamato in caso di elaborazione riuscita del messaggio
messaging:nack(Tack) ‒ viene attivato in tutte le situazioni anomale. Dopo che il task viene restituito, il messaging lo passerà a un altro gestore.

Supponiamo che durante l'elaborazione di tre task si sia verificato un errore complesso: il gestore 1 è andato in crash dopo aver ricevuto il task, senza riuscire a comunicare nulla al punto di scambio. In questo caso, il punto di scambio, trascorso il timeout di ack, passerà il task a un altro gestore. Il gestore 3, per qualche motivo, ha rifiutato il task e ha inviato un nack; di conseguenza, il task è passato a un altro gestore che lo ha completato con successo.

Risultato preliminare

Abbiamo esaminato i principali mattoni dei sistemi distribuiti e ottenuto una base per comprenderne l'applicazione in Erlang/Elixir.

Combinando schemi di base, si possono costruire paradigmi complessi per risolvere i compiti emergenti.

Nella parte conclusiva del ciclo, esamineremo questioni comuni nell'organizzazione dei servizi, nella routinizzazione e nel bilanciamento, e discuteremo anche dell'aspetto pratico della scalabilità e della resilienza dei sistemi.

Fine della seconda parte.

Foto Marius Christensen
Illustrazioni realizzate con websequencediagrams.com

Fonte: habr.com