Byggstenar för distribuerade applikationer. Första tillvägagångssättet

Förr artikeln Vi undersökte de teoretiska grunderna för reaktiv arkitektur. Det är dags att prata om dataflöden, sätt att implementera reaktiva Erlang/Elixir-system och meddelandemönster i dem:

• Begäran-svar
• Request-Chunked svar
• Svar med förfrågan
• Publicera-prenumerera
• Inverterad Publicera-prenumerera
• Uppgiftsfördelning

SOA, MSA och meddelanden

SOA, MSA är systemarkitekturer som definierar reglerna för byggnadssystem, medan meddelandehantering ger primitiver för deras implementering.

Jag vill inte främja den eller den systemarkitekturen. Jag är för att använda de mest effektiva och användbara metoderna för ett specifikt projekt och företag. Vilket paradigm vi än väljer är det bättre att skapa systemblock med ett öga på Unix-sättet: komponenter med minimal anslutning, ansvariga för enskilda enheter. API-metoder utför de enklaste möjliga åtgärderna med entiteter.

Messaging är, som namnet antyder, en meddelandeförmedlare. Dess huvudsakliga syfte är att ta emot och skicka meddelanden. Den ansvarar för gränssnitten för att skicka information, bildandet av logiska kanaler för att överföra information inom systemet, dirigering och balansering samt felhantering på systemnivå.
De meddelanden vi utvecklar försöker inte konkurrera med eller ersätta rabbitmq. Dess huvudfunktioner:

• Distribution.
  Utbytespunkter kan skapas på alla klusternoder, så nära koden som använder dem som möjligt.
• Enkelhet.
  Fokusera på att minimera standardkod och användarvänlighet.
• Bättre prestanda.
  Vi försöker inte upprepa rabbitmqs funktionalitet, utan lyfter bara fram det arkitektoniska och transportlagret, som vi passar in i OTP så enkelt som möjligt, vilket minimerar kostnaderna.
• Flexibilitet.
  Varje tjänst kan kombinera många utbytesmallar.
• Resiliency by design.
• Skalbarhet.
  Meddelanden växer med applikationen. När belastningen ökar kan du flytta utbytespunkterna till enskilda maskiner.

Kommentar. När det gäller kodorganisation är metaprojekt väl lämpade för komplexa Erlang/Elixir-system. All projektkod finns i ett arkiv - ett paraplyprojekt. Samtidigt är mikrotjänster maximalt isolerade och utför enkla operationer som ansvarar för en separat enhet. Med detta tillvägagångssätt är det lätt att underhålla hela systemets API, det är lätt att göra ändringar, det är bekvämt att skriva enhets- och integrationstester.

Systemkomponenterna interagerar direkt eller via en mäklare. Ur ett meddelandeperspektiv har varje tjänst flera livsfaser:

• Serviceinitiering.
  I detta skede konfigureras och startas processen som exekverar tjänsten och dess beroenden.
• Skapa en utbytespunkt.
  Tjänsten kan använda en statisk utbytespunkt specificerad i nodkonfigurationen, eller skapa utbytespunkter dynamiskt.
• Tjänsteregistrering.
  För att tjänsten ska kunna betjäna förfrågningar måste den vara registrerad på bytespunkten.
• Normal funktion.
  Tjänsten ger nyttigt arbete.
• Stänga av.
  Det finns två typer av avstängning möjliga: normal och nödsituation. Under normal drift kopplas tjänsten från växelpunkten och stoppas. I nödsituationer kör meddelanden ett av failover-skripten.

Det ser ganska komplicerat ut, men koden är inte så skrämmande. Kodexempel med kommentarer kommer att ges i analysen av mallar lite senare.

Utbyten

Exchange Point är en meddelandeprocess som implementerar logiken för interaktion med komponenter i meddelandemallen. I alla exempel som presenteras nedan interagerar komponenterna genom utbytespunkter, vars kombination bildar meddelanden.

Mönster för meddelandeutbyte (MEPs)

Globalt kan utbytesmönster delas in i tvåvägs och envägs. Det förra innebär ett svar på ett inkommande meddelande, det senare inte. Ett klassiskt exempel på ett tvåvägsmönster i klient-serverarkitektur är Request-response-mönstret. Låt oss titta på mallen och dess ändringar.

Request-response eller RPC

RPC används när vi behöver få svar från en annan process. Denna process kan köras på samma nod eller på en annan kontinent. Nedan visas ett diagram över interaktionen mellan klient och server via meddelanden.

Eftersom meddelanden är helt asynkron, är växeln för klienten uppdelad i två faser:

1. Skickar en förfrågan

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

   utbyte ‒ unikt namn på utbytespunkten
   ResponseMatchingTag ‒ lokal etikett för bearbetning av svaret. Till exempel när det gäller att skicka flera identiska förfrågningar som tillhör olika användare.
   RequestDefinition - organ för begäran
   HandlerProcess ‒ PID för hanteraren. Denna process kommer att få ett svar från servern.

2. Bearbetar svaret

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

   ResponsePayload - serversvar.

För servern består processen också av 2 faser:

1. Initiering av utbytespunkten
2. Behandling av mottagna förfrågningar

Låt oss illustrera denna mall med kod. Låt oss säga att vi behöver implementera en enkel tjänst som tillhandahåller en enda exakt tidsmetod.

Serverkod

Låt oss definiera tjänstens API i 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{}
}).

Låt oss definiera tjänstekontrollern i 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.

Kundkod

För att skicka en förfrågan till tjänsten kan du anropa API:et för meddelandeförfrågan var som helst i klienten:

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

I ett distribuerat system kan konfigurationen av komponenterna vara mycket olika och vid tidpunkten för begäran kan meddelandeutsändningen ännu inte starta, eller så är tjänstekontrollanten inte redo att betjäna begäran. Därför måste vi kontrollera meddelandesvaret och hantera felfallet.
Efter lyckad sändning kommer klienten att få ett svar eller fel från tjänsten.
Låt oss hantera båda fallen i 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 svar

Det är bäst att undvika att skicka stora meddelanden. Reaktionsförmågan och stabil drift av hela systemet beror på detta. Om svaret på en fråga tar upp mycket minne är det obligatoriskt att dela upp det i delar.

Låt mig ge dig ett par exempel på sådana fall:

• Komponenterna utbyter binär data, såsom filer. Att dela upp responsen i små delar hjälper dig att arbeta effektivt med filer av alla storlekar och undvika minnesspill.
• Listor. Till exempel måste vi välja alla poster från en enorm tabell i databasen och överföra dem till en annan komponent.

Jag kallar dessa svar lokomotiv. Hur som helst är 1024 meddelanden på 1 MB bättre än ett enda meddelande på 1 GB.

I Erlang-klustret får vi en ytterligare fördel - att minska belastningen på utbytespunkten och nätverket, eftersom svaren omedelbart skickas till mottagaren och går förbi utbytespunkten.

Svar med förfrågan

Detta är en ganska sällsynt modifiering av RPC-mönstret för att bygga dialogsystem.

Publicera-prenumerera (datadistributionsträd)

Händelsestyrda system levererar dem till konsumenter så snart data är klar. Således är system mer benägna att använda en push-modell än för en pull- eller poll-modell. Denna funktion låter dig undvika slöseri med resurser genom att ständigt begära och vänta på data.
Bilden visar processen för att distribuera ett meddelande till konsumenter som prenumererar på ett specifikt ämne.

Klassiska exempel på att använda detta mönster är fördelningen av staten: spelvärlden i datorspel, marknadsdata om börser, användbar information i dataflöden.

Låt oss titta på abonnentkoden:

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.

Källan kan anropa funktionen för att publicera ett meddelande på valfri lämplig plats:

messaging:publish_message(Exchange, Key, Message).

utbyte - namnet på bytesplatsen,
Nyckel - routing nyckel
Meddelande - nyttolast

Inverterad Publicera-prenumerera

Genom att expandera pub-sub kan du få ett mönster som är bekvämt för loggning. Uppsättningen av källor och konsumenter kan vara helt olika. Figuren visar ett fall med en konsument och flera källor.

Uppgiftsfördelningsmönster

Nästan varje projekt involverar uppskjutna bearbetningsuppgifter, som att generera rapporter, leverera aviseringar och hämta data från tredje parts system. Genomströmningen av systemet som utför dessa uppgifter kan lätt skalas genom att lägga till hanterare. Allt som återstår för oss är att bilda ett kluster av processorer och jämnt fördela uppgifter mellan dem.

Låt oss titta på de situationer som uppstår med hjälp av exemplet med 3 hanterare. Även vid uppgiftsfördelningen uppstår frågan om rättvis distribution och överflöde av hanterare. Round-robin distribution kommer att ansvara för rättvisa, och för att undvika en situation med översvämning av hanterare kommer vi att införa en begränsning prefetch_limit. Under övergående förhållanden prefetch_limit kommer att förhindra en hanterare från att ta emot alla uppgifter.

Meddelanden hanterar köer och behandlingsprioritet. Processorer får uppgifter när de anländer. Uppgiften kan slutföras framgångsrikt eller misslyckas:

• messaging:ack(Tack) - anropas om meddelandet har bearbetats
• messaging:nack(Tack) - tillkallas i alla akuta situationer. När uppgiften har returnerats kommer meddelanden att skicka den vidare till en annan hanterare.

Antag att ett komplext fel inträffade under bearbetningen av tre uppgifter: processor 1, efter att ha tagit emot uppgiften, kraschade utan att ha tid att rapportera något till utbytespunkten. I det här fallet kommer utbytespunkten att överföra uppgiften till en annan hanterare efter att ack-timeouten har löpt ut. Av någon anledning övergav hanterare 3 uppgiften och skickade nack; som ett resultat överfördes uppgiften också till en annan hanterare som framgångsrikt slutförde den.

Preliminär sammanfattning

Vi har täckt de grundläggande byggstenarna i distribuerade system och fått en grundläggande förståelse för deras användning i Erlang/Elixir.

Genom att kombinera grundläggande mönster kan du bygga komplexa paradigm för att lösa nya problem.

I den sista delen av serien kommer vi att titta på allmänna frågor om att organisera tjänster, routing och balansering, och även prata om den praktiska sidan av skalbarhet och feltolerans hos system.

Slutet på andra delen.

Photo Shoot Marius Christensen
Illustrationer framställda med hjälp av websequencediagrams.com

Källa: will.com