Byggblock för distribuerade applikationer. Första närmandet

Tidigare artikeln Vi har diskuterat de teoretiska grunderna för reaktiv arkitektur. Det är dags att prata om datastreams, implementeringsvägar för reaktiva Erlang/Elixir-system och meddelandeutbyte i dem:

• Begär-svar
• Begär-Chunked Svar
• Svar med Begäran
• Publicera-prenumerera
• Inverterad Publicera-prenumerera
• Uppgiftsdistribution

SOA, MSA och meddelandeutbyte

SOA, MSA – systemarkitekturer som definierar reglerna för systemuppbyggnad, medan meddelandeutbyte tillhandahåller primitiva verktyg för deras implementering.

Jag vill inte propagera för en viss systemarkitektur. Jag förespråkar att använda så effektiva och nyttiga metoder som möjligt för ett specifikt projekt och företag. Oavsett vilken paradigm vi väljer, är det bättre att skapa systemblock med tanke på Unix-way: komponenter med minimal bindning som ansvarar för individuella enheter. API-metoder utför de enklaste åtgärderna med enheterna.

Meddelandeutbyte ‒ som namnet antyder ‒ är en meddelandebroker. Dess huvudsakliga syfte är att ta emot och skicka meddelanden. Den ansvarar för gränssnitten för att skicka information, bildandet av logiska informationsöverföringskanaler inom systemet, routing och belastningsbalansering, samt hantering av fel på systemnivå.
Det utvecklade meddelandeutbytet försöker inte konkurrera med rabbitmq eller ersätta det. Dess huvuddrag är:

• Distribuerbarhet.
  Utbytespunkter kan skapas på alla noder i klustret, så nära koden som möjligt som använder dem.
• Enkelhet.
  Inriktning på minimering av mallkod och användarvänlighet.
• Bästa prestanda.
  Vi försöker inte återge funktionaliteten hos rabbitmq, utan lyfter fram only den arkitektoniska och transportlagret, som vi enkelt kan integrera i OTP och minimera kostnaderna.
• Flexibilitet.
  Varje tjänst kan kombinera många mönster för meddelandeutbyte.
• Felmotstånd, inbäddat i designen.
• Skalbarhet.
  Meddelandeutbytet växer tillsammans med applikationen. Medan belastningen ökar kan man flytta utbytespunkter till separata maskiner.

Anmärkning. Ur kodorganisations synvinkel passar meta-projekt bra för komplexa system på Erlang/Elixir. All kod i projektet finns i ett gemensamt repository ‒ ett paraplyprojekt. Samtidigt är mikrotjänster maximalt isolerade och utför enkla operationer som ansvarar för enskilda entiteter. Med detta tillvägagångssätt är det lätt att underhålla API:et för hela systemet, enkelt att göra ändringar och bekvämt att skriva enhets- och integrationstester.

Systemkomponenter interagerar direkt eller via en broker. Ur messaging-synpunkt har varje tjänst flera livscykelfaser:

• Initiering av tjänsten.
  I detta skede sker konfiguration och start av den exekverande tjänsteprocessen och dess beroenden.
• Skapande av en exchange-punkt.
  Tjänsten kan använda en statisk exchange-punkt som anges i nodens konfiguration, eller skapa exchange-punkter dynamiskt.
• Registrering av tjänsten.
  För att tjänsten ska kunna hantera begärningar måste den registreras vid exchange-punkten.
• Normal funktionalitet.
  Tjänsten utför nytt arbete.
• Avslutning av arbetet.
  Det finns två typer av avslutning: normal och akut. Vid normal avslutning kopplar sig tjänsten bort från exchange-punkten och stängs av. Vid akuta fall utför messaging ett av felhanteringsscenarierna.

Det ser ganska komplicerat ut, men i koden är det inte så skrämmande. Kodexempel med kommentarer kommer att presenteras i analysen av mönstren lite senare.

Exchanges

En exchange-punkt är en messaging-process som implementerar logiken för interaktion med komponenter inom ramen för meddelandeutbytesmönstret. I alla exempel som presenteras nedan interagerar komponenterna via exchange-punkter, vars kombination utgör messaging.

Message exchange patterns (MEPs)

Globalt kan utbytesmallar delas in i tvåvägs och envägs. De första innebär ett svar på det mottagna meddelandet, de andra gör det inte. Ett klassiskt exempel på en tvåvägs-mall i en klient-server-arkitektur är Request-response mönstret. Låt oss granska mönstret och dess modifieringar.

Request–response eller RPC

RPC används när vi behöver få ett svar från en annan process. Denna process kan köras på samma nod eller befinna sig på en annan kontinent. Nedan visas en skiss över interaktionen mellan klienten och servern via messaging.

Eftersom messaging är helt asynkront, delas utbytet för klienten in i två faser:

1. Skicka begäran

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

   Exchange ‒ ett unikt namn för utbytespunkten
   ResponseMatchingTag ‒ en lokal etikett för att bearbeta svaret. Till exempel, i fall av att skicka flera samma förfrågningar som tillhör olika användare.
   RequestDefinition ‒ kroppen av förfrågan
   HandlerProcess ‒ PID för hanteraren. Denna process kommer att få svaret från servern.

2. Bearbetning av svaret

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

   ResponsePayload ‒ svaret från servern.

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

1. Initiering av utbytespunkten
2. Bearbetning av inkommande förfrågningar

Låt oss illustrera detta mönster med kod. Anta att vi behöver implementera en enkel tjänst som tillhandahåller en enda metod för exakt tid.

Serverkod

Vi flyttar definitionen av tjänstens API till api.hrl:

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

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

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

Vi definierar tjänstens kontroller i time_controller.erl

%% Exemplet visar endast relevant kod. Genom att infoga det i mönstret gen_server kan vi få en fungerande tjänst.

%% initiering av gen_server
init(Args) ->
  %% anslutning till utbytespunkten
  messaging:monitor_exchange(req_resp, ?EXCHANGE, default, self())
  {ok, #{}}.

%% hantering av händelsen för bortfall av kontakt med utbytespunkten. Denna händelse skickas också om utbytespunkten ännu inte har startat.
handle_info(#exchange_die{exchange = ?EXCHANGE}, State) ->
  erlang:send(self(), monitor_exchange),
  {noreply, State};

%% hantering av 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};

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

Kundkod

För att skicka en begäran till tjänsten kan messaging begäran API anropas var som helst i klienten:

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

I ett distribuerat system kan konfigurationen av komponenter vara mycket varierande, och när begäran görs kan messaging ännu inte ha startat, eller så kan tjänstkontrollern inte vara redo att hantera begäran. Därför måste vi kontrollera svaret från messaging och hantera avvisningsfallet.
Efter en framgångsrik sändning kommer klienten att få ett svar eller ett fel från tjänsten.
Vi hanterar 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};

Begär-Chunked Svar

Det är bäst att undvika att skicka stora meddelanden. Detta påverkar responsen och stabiliteten hos hela systemet. Om svaret på en begäran tar mycket minne, är det nödvändigt att dela upp det i delar.

Jag ska ge några exempel på sådana fall:

• Komponenter byter binära data, exempelvis filer. Att dela upp svaret i små delar gör det möjligt att effektivt hantera filer av alla storlekar utan att få minnesöverfull.
• Listor. Till exempel behöver vi hämta alla poster från en stor tabell i databasen och skicka dem till en annan komponent.

Jag kallar sådana svar för en lokomotiv. I vilket fall som helst, 1024 meddelanden på 1 MB är bättre än ett enda meddelande som är 1 GB.

I en Erlang-kluster får vi ytterligare fördelar - minskad belastning på utbytespunkten och nätverket, eftersom svaren skickas direkt till mottagaren utan att gå via utbytespunkten.

Svar med Begäran

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

Publicera-prenumerera (datadistributionsstruktur)

Händelseorienterade system levererar data till konsumenterna så snart de är redo. På så sätt är systemen mer benägna att använda en push-modell än en pull eller poll. Denna egenskap förhindrar slöseri med resurser genom att ständigt begära och vänta på data.
På bilden visas processen för distribution av meddelanden till konsumenter som prenumererat på ett visst ämne.

Klassiska exempel på användning av detta mönster är distribution av tillstånd: spelvärldar i datorspel, marknadsdata på börser, användbar information i dataflöden.

Låt oss titta på koden för prenumeranten:

init(_Args) ->
  %% prenumerera på utbytet, nyckel = key
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {ok, #{}}.

handle_info(#exchange_die{exchange = ?SUBSCRIPTION}, State) ->
  %% om utbytet inte är tillgängligt, försök att återansluta
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {noreply, State};

%% bearbeta inkommande meddelanden
handle_info(#'$msg'{exchange = ?SUBSCRIPTION, message = Msg}, State) ->
  ?debugVal(Msg),
  {noreply, State};

%% vid stopp av konsumenten - koppla bort från utbytet
terminate(_Reason, _State) ->
  messaging:unsubscribe(?SUBSCRIPTION, key, tag, self()),
  ok.

Källa kan anropa publiceringsfunktionen var som helst:

messaging:publish_message(Exchange, Key, Message).

Exchange ‒ namnet på utbytespunkten,
Nyckel ‒ routningsnyckel
Message ‒ nyttolast

Inverterad Publicera-prenumerera

Genom att rulla ut pub-sub kan man få ett mönster som är bekvämt för loggning. Mängden källor och konsumenter kan variera helt. Bilden visar ett fall med en konsument och många källor.

Uppgiftfördelningsmönster

I nästan varje projekt uppstår uppgifter för fördröjd behandling, såsom rapportgenerering, aviseringar och datainsamling från externa system. Kapaciteten hos systemet som utför dessa uppgifter kan enkelt skalas genom att lägga till bearbetare. Det enda vi behöver göra är att skapa en kluster av bearbetare och jämnt fördela uppgifterna mellan dem.

Låt oss överväga situationer med 3 bearbetare. Redan i automatiseringssteget uppstår frågan om rättvis fördelning och överbelastning av bearbetarna. Rättvisan kommer att säkerställas genom round-robin fördelning, och för att undvika överbelastningssituationen kommer vi att införa begränsningen prefetch_limit. I övergångstillstånd prefetch_limit kan en bearbetare inte få alla uppgifter.

Messaging hanterar köer och prioriteringar för behandling. Bearbetarna får uppgifter i takt med att de kommer in. Uppgiftsutförandet kan avslutas antingen med framgång eller med ett fel:

• messaging:ack(Tack) ‒ anropas vid framgångsrik behandling av meddelandet
• messaging:nack(Tack) ‒ anropas vid alla onormala situationer. Efter att uppgiften har återlämnats kommer messaging att vidarebefordra den till en annan bearbetare.

Anta att det inträffade ett komplicerat fel vid bearbetning av tre uppgifter: bearbetare 1 kraschade efter att ha fått uppgiften utan att hinna meddela utbytespunkten. I detta fall kommer utbytespunkten, efter att ack timeout har löpt ut, att överföra uppgiften till en annan bearbetare. Bearbetare 3 avstod från uppgiften av någon anledning och skickade nack, vilket resulterade i att uppgiften även överfördes till en annan bearbetare som framgångsrikt genomförde den.

Preliminärt resultat

Vi har behandlat de grundläggande byggstenarna i distribuerade system och fått en grundläggande förståelse för deras tillämpningar i Erlang/Elixir.

Genom att kombinera grundläggande mallar kan man bygga komplexa paradigmer för att lösa uppkommande problem.

I den avslutande delen av cykeln kommer vi att diskutera allmänna frågor om organisation av tjänster, ruttningssystem och belastningsbalansering, samt prata om de praktiska aspekterna av skalbarhet och motståndskraft hos system.

Slutet av del två.

Foto Marius Christensen
Illustrationer skapade med hjälp av websequencediagrams.com

Källa: habr.com