Bouwstenen van gedistribueerde applicaties. Eerste benadering

In het verleden artikel hebben we de theoretische basis van reactieve architectuur behandeld. Het is tijd om te praten over datastromen, manieren om reactieve Erlang/Elixir-systemen te implementeren en de message exchange patronen binnen deze systemen:

• Request-response
• Request-Chunked Response
• Response with Request
• Publish-subscribe
• Inverted Publish-subscribe
• Taakverdeling

SOA, MSA en messaging

SOA, MSA – systeemarchitecturen die de regels voor het bouwen van systemen definiëren, terwijl messaging de primitieve elementen voor hun implementatie levert.

Ik wil geen bepaalde systeemarchitectuur propageren. Ik pleit voor het toepassen van maximaal effectieve en nuttige praktijken voor een specifiek project en bedrijf. Welke paradigma we ook kiezen, het is beter om systeembouwblokken te creëren met het Unix-principe in gedachten: componenten met minimale koppeling, verantwoordelijk voor afzonderlijke entiteiten. API-methoden voeren de eenvoudigste acties uit met entiteiten.

Messaging ‒ zoals de naam al impliceert ‒ is een berichtenbroker. Het belangrijkste doel is om berichten te ontvangen en te versturen. Het zorgt voor de interfaces voor het verzenden van informatie, het opzetten van logische communicatiewegen binnen het systeem, routering en load balancing, evenals het omgaan met systeemfouten.
De ontwikkelde messaging probeert niet te concurreren met rabbitmq of het te vervangen. De belangrijkste kenmerken zijn:

• Gedispliceerdheid.
  Punten voor uitwisseling kunnen op alle knooppunten van het cluster worden gecreëerd, zo dicht mogelijk bij de code die ze gebruikt.
• Eenvoud.
  Gerichtheid op de minimalisering van template-code en gebruiksgemak.
• Betere prestaties.
  We proberen niet de functionaliteit van rabbitmq te herhalen, maar belichten alleen de architecturale en transportlaag, die we zo eenvoudig mogelijk integreren in OTP, waarbij we de kosten minimaliseren.
• Flexibiliteit.
  Elke service kan meerdere exchange patronen combineren.
• Fouttolerantie, ingebouwd in het ontwerp.
• Schaalbaarheid.
  Messaging groeit samen met de applicatie. Naarmate de belasting toeneemt, kunnen uitwisselingspunten naar aparte machines worden verplaatst.

Opmerking. Vanuit het oogpunt van codeorganisatie zijn meta-projecten zeer geschikt voor complexe systemen op Erlang/Elixir. Alle projectcode bevindt zich in één repository – het overkoepelende project. Microservices zijn hierbij maximaal geïsoleerd en voeren eenvoudige bewerkingen uit, verantwoordelijk voor afzonderlijke entiteiten. Met deze aanpak is het eenvoudig om de API van het hele systeem te onderhouden, wijzigingen aan te brengen en unit- en integratietests te schrijven.

De systeemcomponenten communiceren rechtstreeks of via een broker. Vanuit het oogpunt van messaging heeft elke service meerdere levensfasen:

• Initialisatie van de service.
  In deze fase vindt de configuratie en de uitvoering van het serviceproces en zijn afhankelijkheden plaats.
• Creëren van een exchange.
  De service kan gebruikmaken van een statische exchange die in de configuratie van de node is ingesteld, of dynamisch exchanges creëren.
• Registratie van de service.
  Om de service verzoeken te laten afhandelen, moet deze op de exchange worden geregistreerd.
• Normale werking.
  De service verricht nuttig werk.
• Beëindiging van de werking.
  Er zijn 2 soorten beëindiging: normaal en abnormaal. Bij een normale beëindiging wordt de service losgekoppeld van de exchange en gestopt. In geval van abnormale situaties voert messaging een van de foutafhandelingsscenario's uit.

Het lijkt best ingewikkeld, maar in de code is het niet zo erg. Voorbeelden van code met opmerkingen zullen later bij de analyse van de templates worden gepresenteerd.

Exchanges

Een exchange is een messagingproces dat de logica van interactie met componenten binnen het messagingtemplate implementeert. In alle hieronder gepresenteerde voorbeelden communiceren de componenten via exchanges, waarvan de combinatie messaging vormt.

Message exchange patterns (MEPs)

Globaal kunnen de templates voor berichtuitwisseling worden onderverdeeld in bidirectionele en unidirectionele. De eerste impliceren een antwoord op het ontvangen bericht, de laatste niet. Een klassiek voorbeeld van een bidirectioneel template in client-serverarchitecturen is het Request-response template. Laten we het template en zijn modificaties bekijken.

Request–response of RPC

RPC wordt gebruikt wanneer we een antwoord van een ander proces nodig hebben. Dit proces kan op dezelfde node worden uitgevoerd of zich op een ander continent bevinden. Hieronder wordt het schema van de interactie tussen de client en de server via messaging.

Aangezien messaging volledig asynchroon is, is de uitwisseling voor de client verdeeld in 2 fasen:

1. Verzending van het verzoek

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

   Exchange ‒ unieke naam van het uitwisselingspunt
   ResponseMatchingTag ‒ lokale tag voor het verwerken van het antwoord. Bijvoorbeeld bij het verzenden van meerdere identieke aanvragen, die van verschillende gebruikers zijn.
   RequestDefinition ‒ lichaam van de aanvraag
   HandlerProcess ‒ PID van de handler. Dit proces ontvangt de reactie van de server.

2. Verwerken van het antwoord

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

   ResponsePayload ‒ reactie van de server.

Voor de server bestaat het proces ook uit 2 fasen:

1. Initialisatie van het uitwisselingspunt
2. Verwerking van binnenkomende aanvragen

Laten we deze sjabloon met code illustreren. Stel dat we een eenvoudige service moeten implementeren die een enkele methode voor de exacte tijd biedt.

Servercode

Laten we de API-definitie van de service in api.hrl plaatsen:

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

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

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

Laten we de servicecontroller definiëren in time_controller.erl

%% In het voorbeeld wordt alleen essentiële code getoond. Door het in het gen_server-sjabloon in te voegen, kan een werkende service worden verkregen.

%% initialisatie van gen_server
init(Args) ->
  %% verbinding maken met het uitwisselingspunt
  messaging:monitor_exchange(req_resp, ?EXCHANGE, default, self())
  {ok, #{}}.

%% verwerking van het evenement van het verlies van verbinding met het uitwisselingspunt. Dit evenement komt ook binnen als het uitwisselingspunt nog niet is gestart.
handle_info(#exchange_die{exchange = ?EXCHANGE}, State) ->
  erlang:send(self(), monitor_exchange),
  {noreply, State};

%% verwerking van de 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};

%% beëindiging van de werking van gen_server
terminate(_Reason, _State) ->
  messaging:demonitor_exchange(req_resp, ?EXCHANGE, default, self()),
  ok.

Klantcode

Om een aanvraag naar de service te sturen, kan op elk moment in de klant de messaging aanvraag-API worden aangeroepen:

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

In een gedistribueerd systeem kan de configuratie van componenten zeer verschillend zijn en op het moment van de aanvraag kan messaging nog niet zijn gestart, of de servicecontroller is mogelijk niet klaar om de aanvraag te verwerken. Daarom moeten we de reactie van messaging controleren en omgaan met het geval van een fout.
Na een succesvolle verzending ontvangt de klant een reactie of fout van de service.
Laten we beide gevallen verwerken 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

Het is beter om te voorkomen dat enorme berichten worden verzonden. Dit beïnvloedt de responstijd en de stabiliteit van het hele systeem. Als het antwoord op een verzoek veel geheugen vereist, is het opsplitsen in delen verplicht.

Hier zijn een paar voorbeelden van dergelijke gevallen:

• Componenten wisselen binaire gegevens uit, zoals bestanden. Het opsplitsen van het antwoord in kleinere delen helpt efficiënt met bestanden van elke grootte om te gaan en voorkomt geheugenoverloop.
• Lijsten. Bijvoorbeeld, we moeten alle records uit een enorme tabel in de database selecteren en deze naar een andere component verzenden.

Ik noem dergelijke antwoorden een 'stoomlocomotief'. In ieder geval zijn 1024 berichten van 1 MB beter dan één enkele boodschap van 1 GB.

In een Erlang-cluster behalen we een extra voordeel - de belasting op het uitwisselpunt en het netwerk verminderen, aangezien de antwoorden rechtstreeks naar de ontvanger worden gestuurd, waarbij het uitwisselpunt wordt omzeild.

Response with Request

Dit is een vrij zeldzame wijziging van het RPC-patroon voor het bouwen van dialoogsystemen.

Publiceer-abonneer (gegevensdistributiebomen)

Evenementgeorieneteerde systemen leveren gegevens aan consumenten zodra deze beschikbaar zijn. Daardoor zijn systemen meer geneigd naar een push-model dan naar pull of poll. Deze eigenschap voorkomt verspilling van middelen door constant op gegevens te wachten en deze op te vragen.
De afbeelding toont het proces van het verspreiden van berichten naar consumenten die zich op een bepaald onderwerp hebben geabonneerd.

Klassieke voorbeelden van het gebruik van dit patroon zijn het verspreiden van de status: van de spelwereld in computerspellen, marktgegevens op beurzen, en nuttige informatie in datafeeds.

Laten we de code van de abonnee bekijken:

init(_Args) ->
  %% we abonneren ons op het uitwisselpunt, sleutel = key
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {ok, #{}}.

handle_info(#exchange_die{exchange = ?SUBSCRIPTION}, State) ->
  %% als het uitwisselpunt niet beschikbaar is, proberen we opnieuw verbinding te maken
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {noreply, State};

%% verwerk de binnenkomende berichten
handle_info(#'$msg'{exchange = ?SUBSCRIPTION, message = Msg}, State) ->
  ?debugVal(Msg),
  {noreply, State};

%% bij stop van de ontvanger - ontkoppelen van het uitwisselpunt
terminate(_Reason, _State) ->
  messaging:unsubscribe(?SUBSCRIPTION, key, tag, self()),
  ok.

De bron kan de functie voor het publiceren van een bericht op elke gewenste plek aanroepen:

messaging:publish_message(Exchange, Key, Message).

Exchange ‒ de naam van het uitwisselpunt,
Key ‒ het routeringssleutel
Message ‒ de payload

Inverted Publish-subscribe

Door pub-sub te implementeren, kan een patroon worden verkregen dat handig is voor logging. De set van bronnen en consumenten kan volledig verschillend zijn. In de afbeelding wordt een geval weergegeven met één consument en meerdere bronnen.

Taakverdelingspatroon

In bijna elk project komen taken voor die uitgestelde verwerking vereisen, zoals het genereren van rapporten, het bezorgen van meldingen en het ophalen van gegevens uit externe systemen. De capaciteit van het systeem dat deze taken uitvoert, kan eenvoudig worden opgeschaald door verwerkers toe te voegen. Het enige wat we hoeven te doen, is een cluster van verwerkers te vormen en de taken gelijkmatig over hen te verdelen.

Laten we de situaties bekijken aan de hand van 3 verwerkers. Al tijdens de taakverdeling rijst de vraag naar de rechtvaardigheid van de verdeling en de overbelasting van verwerkers. De rechtvaardigheid zal worden gewaarborgd door round-robin verdeling, en om overbelasting van de verwerkers te voorkomen, introduceren we een beperking prefetch_limit. In overgangstoestanden prefetch_limit stof niet toestaan dat één verwerker alle taken krijgt.

Messaging beheert de wachtrijen en de prioriteit van verwerking. Verwerkers ontvangen taken naarmate ze binnenkomen. De uitvoering van een taak kan succesvol worden afgerond of mislukken:

• messaging:ack(Tack) ‒ wordt aangeroepen in het geval van succesvolle verwerking van het bericht
• messaging:nack(Tack) ‒ wordt aangeroepen in alle noodgevallen. Na terugzending van de taak, zal messaging deze naar een andere verwerker doorgeven.

Stel dat er tijdens de verwerking van drie taken een complexe storing optreedt: verwerker 1 viel uit na het ontvangen van de taak, zonder iets naar het uitwisselpunt te communiceren. In dit geval zal het uitwisselpunt na het verstrijken van de ack timeout de taak naar een andere verwerker doorgeven. Verwerker 3 heeft om een bepaalde reden de taak afgewezen en een nack gestuurd, waardoor de taak ook naar een andere verwerker ging die deze succesvol voltooide.

Tussentijdse conclusie

We hebben de belangrijkste bouwstenen van gedistribueerde systemen behandeld en een basisbegrip van hun toepassing in Erlang/Elixir gekregen.

Door basispatronen te combineren, kunnen complexe paradigma's worden opgebouwd voor het oplossen van de opkomende taken.

In het laatste deel van de cyclus bespreken we algemene vragen over de organisatie van diensten, routering en load balancing, en we zullen ook de praktische kant van schaalbaarheid en fouttolerantie van systemen bespreken.

Einde van het tweede deel.

Foto Marius Christensen
Illustraties zijn gemaakt met behulp van websequencediagrams.com

Bron: habr.com