Bouwstenen van gedistribueerde applicaties. Eerste aanpak

In het verleden статье We onderzochten de theoretische grondslagen van reactieve architectuur. Het is tijd om te praten over datastromen, manieren om reactieve Erlang/Elixir-systemen en berichtpatronen daarin te implementeren:

• Aanvraag antwoord
• Verzoek-gefragmenteerd antwoord
• Reactie met verzoek
• Publiceren-abonneren
• Omgekeerd publiceren-abonneren
• Taakverdeling

SOA, MSA en berichtenuitwisseling

SOA en MSA zijn systeemarchitecturen die de regels definiëren voor het bouwen van systemen, terwijl berichtenuitwisseling primitieven biedt voor de implementatie ervan.

Ik wil deze of gene systeemarchitectuur niet promoten. Ik ben voor het gebruik van de meest effectieve en bruikbare praktijken voor een specifiek project en bedrijf. Welk paradigma we ook kiezen, het is beter om systeemblokken te creëren met het oog op de Unix-manier: componenten met minimale connectiviteit, verantwoordelijk voor individuele entiteiten. API-methoden voeren de eenvoudigst mogelijke acties uit met entiteiten.

Messaging is, zoals de naam al doet vermoeden, een berichtenmakelaar. Het voornaamste doel is het ontvangen en verzenden van berichten. Het is verantwoordelijk voor de interfaces voor het verzenden van informatie, de vorming van logische kanalen voor het verzenden van informatie binnen het systeem, routering en balancering, evenals foutafhandeling op systeemniveau.
De boodschap die we ontwikkelen is niet bedoeld om te concurreren met of om konijnmq te vervangen. De belangrijkste kenmerken:

• Verdeling.
  Uitwisselingspunten kunnen op alle clusterknooppunten worden aangemaakt, zo dicht mogelijk bij de code die ze gebruikt.
• Eenvoud.
  Focus op het minimaliseren van standaardcode en gebruiksgemak.
• Betere prestatie.
  We proberen niet de functionaliteit van konijnmq te herhalen, maar benadrukken alleen de architecturale en transportlaag, die we zo eenvoudig mogelijk in het OTP passen, waardoor de kosten worden geminimaliseerd.
• Flexibiliteit.
  Elke dienst kan vele uitwisselingssjablonen combineren.
• Veerkracht door ontwerp.
• Schaalbaarheid.
  Berichten groeien mee met de applicatie. Naarmate de belasting toeneemt, kunt u de wisselpunten naar individuele machines verplaatsen.

Opmerking. In termen van code-organisatie zijn metaprojecten zeer geschikt voor complexe Erlang/Elixir-systemen. Alle projectcode bevindt zich in één repository: een overkoepelend project. Tegelijkertijd zijn microservices maximaal geïsoleerd en voeren ze eenvoudige handelingen uit die verantwoordelijk zijn voor een afzonderlijke entiteit. Met deze aanpak is het eenvoudig om de API van het hele systeem te onderhouden, is het gemakkelijk om wijzigingen aan te brengen en is het handig om unit- en integratietests te schrijven.

De systeemcomponenten werken rechtstreeks of via een makelaar met elkaar samen. Vanuit een berichtenperspectief kent elke dienst verschillende levensfasen:

• Service-initialisatie.
  In dit stadium worden het proces dat de service uitvoert en de afhankelijkheden ervan geconfigureerd en gestart.
• Het creëren van een uitwisselingspunt.
  De service kan een statisch uitwisselingspunt gebruiken dat is opgegeven in de knooppuntconfiguratie, of dynamisch uitwisselingspunten creëren.
• Dienstregistratie.
  Om ervoor te zorgen dat de dienst verzoeken kan verwerken, moet deze op het uitwisselingspunt worden geregistreerd.
• Normaal functioneren.
  De dienst levert nuttig werk op.
• Afsluiten.
  Er zijn 2 soorten uitschakeling mogelijk: normaal en nood. Tijdens normaal bedrijf wordt de dienst losgekoppeld van het wisselpunt en stopt deze. In noodsituaties voert messaging een van de failover-scripts uit.

Het ziet er behoorlijk ingewikkeld uit, maar de code is niet zo eng. Codevoorbeelden met commentaar zullen iets later worden gegeven bij de analyse van sjablonen.

Exchanges

Exchange Point is een berichtenproces dat de logica van interactie met componenten binnen de berichtensjabloon implementeert. In alle onderstaande voorbeelden werken de componenten samen via uitwisselingspunten, waarvan de combinatie berichtenuitwisseling vormt.

Patronen voor berichtenuitwisseling (EP-leden)

Wereldwijd kunnen uitwisselingspatronen worden onderverdeeld in tweerichtings- en eenrichtingsverkeer. De eerste impliceren een reactie op een binnenkomend bericht, de laatste niet. Een klassiek voorbeeld van een tweerichtingspatroon in de client-serverarchitectuur is het Request-response-patroon. Laten we eens kijken naar de sjabloon en de wijzigingen ervan.

Verzoek-antwoord of RPC

RPC wordt gebruikt wanneer we een reactie van een ander proces moeten ontvangen. Dit proces kan op hetzelfde knooppunt worden uitgevoerd of zich op een ander continent bevinden. Hieronder ziet u een diagram van de interactie tussen client en server via berichtenuitwisseling.

Omdat de berichtenuitwisseling volledig asynchroon is, is de uitwisseling voor de klant verdeeld in 2 fasen:

1. Een verzoek verzenden

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

   uitwisseling ‒ unieke naam van het wisselpunt
   ReactieMatchingTag ‒ lokaal label voor het verwerken van het antwoord. Bijvoorbeeld in het geval van het verzenden van meerdere identieke verzoeken van verschillende gebruikers.
   Aanvraagdefinitie - verzoekorgaan
   HandlerProces ‒ PID van de handler. Dit proces ontvangt een antwoord van de server.

2. Het verwerken van de reactie

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

   ReactiePayload - serverreactie.

Voor de server bestaat het proces eveneens uit 2 fases:

1. Initialiseren van het uitwisselingspunt
2. Verwerken van ontvangen aanvragen

Laten we deze sjabloon illustreren met code. Laten we zeggen dat we een eenvoudige service moeten implementeren die één enkele exacte tijdmethode biedt.

Servercode

Laten we de service-API definiëren 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{}
}).

Laten we de servicecontroller definiëren 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.

Klantcode

Om een ​​verzoek naar de service te sturen, kunt u de API voor berichtenverzoeken overal in de client aanroepen:

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

In een gedistribueerd systeem kan de configuratie van componenten heel verschillend zijn en op het moment van het verzoek start de berichtgeving mogelijk nog niet, of is de servicecontroller niet gereed om aan het verzoek te voldoen. Daarom moeten we het berichtantwoord controleren en het foutgeval afhandelen.
Na succesvolle verzending ontvangt de klant een reactie of foutmelding van de dienst.
Laten we beide gevallen behandelen 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};

Verzoek-gefragmenteerd antwoord

Het is het beste om te voorkomen dat u grote berichten verzendt. Het reactievermogen en de stabiele werking van het hele systeem zijn hiervan afhankelijk. Als het antwoord op een query veel geheugen in beslag neemt, is het opsplitsen ervan in delen verplicht.

Ik zal u een paar voorbeelden geven van dergelijke gevallen:

• De componenten wisselen binaire gegevens uit, zoals bestanden. Door de respons in kleine delen op te delen, kunt u efficiënt werken met bestanden van elke grootte en geheugenoverflows voorkomen.
• Advertenties. We moeten bijvoorbeeld alle records uit een enorme tabel in de database selecteren en deze naar een ander onderdeel overbrengen.

Ik noem deze reacties locomotief. Sowieso zijn 1024 berichten van 1 MB beter dan een enkel bericht van 1 GB.

In het Erlang-cluster krijgen we een bijkomend voordeel: het verminderen van de belasting van het uitwisselingspunt en het netwerk, omdat antwoorden onmiddellijk naar de ontvanger worden verzonden, waarbij het uitwisselingspunt wordt omzeild.

Reactie met verzoek

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

Publiceren-abonneren (gegevensdistributieboom)

Gebeurtenisgestuurde systemen leveren ze aan consumenten zodra de gegevens gereed zijn. Systemen zijn dus gevoeliger voor een push-model dan voor een pull- of poll-model. Met deze functie kunt u voorkomen dat u bronnen verspilt door voortdurend gegevens op te vragen en erop te wachten.
De figuur toont het proces van het verspreiden van een bericht naar consumenten die op een specifiek onderwerp zijn geabonneerd.

Klassieke voorbeelden van het gebruik van dit patroon zijn de verdeling van de staat: de spelwereld in computerspellen, marktgegevens over beurzen, nuttige informatie in datafeeds.

Laten we eens kijken naar de abonneecode:

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.

De bron kan de functie aanroepen om een ​​bericht op elke geschikte plaats te publiceren:

messaging:publish_message(Exchange, Key, Message).

uitwisseling - naam van het uitwisselingspunt,
sleutel - routeringssleutel
Bericht - lading

Omgekeerd publiceren-abonneren

Door pub-sub uit te breiden, kunt u een patroon krijgen dat handig is voor loggen. De set van bronnen en consumenten kan totaal verschillend zijn. De figuur toont een casus met één consument en meerdere bronnen.

Patroon voor taakverdeling

Bijna elk project brengt uitgestelde verwerkingstaken met zich mee, zoals het genereren van rapporten, het leveren van meldingen en het ophalen van gegevens uit systemen van derden. De doorvoer van het systeem dat deze taken uitvoert, kan eenvoudig worden geschaald door handlers toe te voegen. Het enige dat ons nog rest is het vormen van een cluster van processors en het gelijkmatig verdelen van de taken daartussen.

Laten we eens kijken naar de situaties die zich voordoen aan de hand van het voorbeeld van 3 handlers. Zelfs in het stadium van de taakverdeling rijst de vraag naar de eerlijkheid van de verdeling en de overvloed aan behandelaars. Round-robin-distributie zal verantwoordelijk zijn voor eerlijkheid, en om een ​​situatie van overstroming van afhandelaars te voorkomen, zullen we een beperking invoeren prefetch_limit. In voorbijgaande omstandigheden prefetch_limit voorkomt dat één handler alle taken ontvangt.

Berichten beheren wachtrijen en verwerkingsprioriteit. Verwerkers ontvangen taken zodra ze binnenkomen. De taak kan met succes worden voltooid of mislukken:

• messaging:ack(Tack) - gebeld als het bericht succesvol is verwerkt
• messaging:nack(Tack) - gebeld in alle noodsituaties. Zodra de taak is geretourneerd, wordt deze door de berichtendienst doorgegeven aan een andere afhandelaar.

Stel dat er een complexe fout optreedt tijdens het verwerken van drie taken: processor 1 crasht na ontvangst van de taak zonder tijd te hebben gehad om iets aan het centrale punt te melden. In dit geval zal het uitwisselingspunt de taak overdragen aan een andere handler nadat de ack-time-out is verstreken. Om de een of andere reden verliet handler 3 de taak en stuurde een nack; als resultaat werd de taak ook overgedragen aan een andere handler die deze met succes voltooide.

Voorlopig resultaat

We hebben de basisbouwstenen van gedistribueerde systemen besproken en basiskennis gekregen van hun gebruik in Erlang/Elixir.

Door basispatronen te combineren, kun je complexe paradigma’s bouwen om opkomende problemen op te lossen.

In het laatste deel van de serie zullen we kijken naar algemene kwesties rond het organiseren van services, routering en balancering, en ook praten over de praktische kant van schaalbaarheid en fouttolerantie van systemen.

Einde van het tweede deel.

foto Marius Kristensen
Illustraties gemaakt met websequencediagrams.com

Bron: www.habr.com