Byggesteiner av distribuerte applikasjoner. Første tilnærming

I den siste artikkel Vi undersøkte det teoretiske grunnlaget for reaktiv arkitektur. Det er på tide å snakke om dataflyter, måter å implementere reaktive Erlang/Elixir-systemer og meldingsmønstre i dem:

• Forespørsel-svar
• Request-Chunked Response
• Svar med forespørsel
• Publiser-abonner
• Invertert Publiser-abonner
• Oppgavefordeling

SOA, MSA og meldinger

SOA, MSA er systemarkitekturer som definerer reglene for byggesystemer, mens meldingstjenester gir primitiver for implementeringen av dem.

Jeg ønsker ikke å fremme denne eller den systemarkitekturen. Jeg er for å bruke de mest effektive og nyttige praksisene for et spesifikt prosjekt og virksomhet. Uansett hvilket paradigme vi velger, er det bedre å lage systemblokker med øye på Unix-veien: komponenter med minimal tilkobling, ansvarlig for individuelle enheter. API-metoder utfører de enklest mulige handlingene med entiteter.

Meldingstjenester er, som navnet tilsier, en meldingsmegler. Hovedformålet er å motta og sende meldinger. Den er ansvarlig for grensesnittene for sending av informasjon, dannelse av logiske kanaler for overføring av informasjon i systemet, ruting og balansering, samt feilhåndtering på systemnivå.
Meldingen vi utvikler prøver ikke å konkurrere med eller erstatte rabbitmq. Hovedfunksjonene:

• Fordeling.
  Utvekslingspunkter kan opprettes på alle klyngenoder, så nærme som mulig koden som bruker dem.
• Enkelhet.
  Fokuser på å minimere standardkode og brukervennlighet.
• Bedre ytelse.
  Vi prøver ikke å gjenta funksjonaliteten til rabbitmq, men fremhever bare det arkitektoniske og transportmessige laget, som vi passer inn i OTP så enkelt som mulig, og minimerer kostnadene.
• Fleksibilitet.
  Hver tjeneste kan kombinere mange utvekslingsmaler.
• Spenst ved design.
• Skalerbarhet.
  Meldinger vokser med applikasjonen. Etter hvert som belastningen øker, kan du flytte byttepoengene til individuelle maskiner.

Merk. Når det gjelder kodeorganisering, er metaprosjekter godt egnet for komplekse Erlang/Elixir-systemer. All prosjektkode er plassert i ett depot - et paraplyprosjekt. Samtidig er mikrotjenester maksimalt isolert og utfører enkle operasjoner som er ansvarlige for en egen enhet. Med denne tilnærmingen er det enkelt å vedlikeholde API-en til hele systemet, det er enkelt å gjøre endringer, det er praktisk å skrive enhets- og integrasjonstester.

Systemkomponentene samhandler direkte eller gjennom en megler. Fra et meldingsperspektiv har hver tjeneste flere livsfaser:

• Tjenesteinitialisering.
  På dette stadiet blir prosessen som utfører tjenesten og dens avhengigheter konfigurert og lansert.
• Opprette et byttepunkt.
  Tjenesten kan bruke et statisk utvekslingspunkt spesifisert i nodekonfigurasjonen, eller opprette utvekslingspunkter dynamisk.
• Tjenesteregistrering.
  For at tjenesten skal betjene forespørsler, må den være registrert på byttepunktet.
• Normal funksjon.
  Tjenesten produserer nyttig arbeid.
• Gjennomføring av arbeid.
  Det er 2 typer avstengning mulig: normal og nødstilfelle. Ved normal drift kobles tjenesten fra utvekslingspunktet og stopper. I nødssituasjoner kjører meldingstjenester ett av failover-skriptene.

Det ser ganske komplisert ut, men koden er ikke så skummel. Kodeeksempler med kommentarer vil bli gitt i analysen av maler litt senere.

Børser

Exchange point er en meldingsprosess som implementerer logikken for interaksjon med komponenter i meldingsmalen. I alle eksemplene som presenteres nedenfor, samhandler komponentene gjennom utvekslingspunkter, hvor kombinasjonen danner meldinger.

Meldingsutvekslingsmønstre (MEPs)

Globalt kan utvekslingsmønstre deles inn i toveis og enveis. Førstnevnte innebærer et svar på en innkommende melding, sistnevnte ikke. Et klassisk eksempel på et toveismønster i klient-server-arkitektur er Request-response-mønsteret. La oss se på malen og dens modifikasjoner.

Forespørsel-svar eller RPC

RPC brukes når vi trenger å motta svar fra en annen prosess. Denne prosessen kan kjøre på samme node eller lokalisert på et annet kontinent. Nedenfor er et diagram over interaksjonen mellom klient og server via meldinger.

Siden meldinger er helt asynkrone, er sentralen for klienten delt inn i 2 faser:

1. Sender forespørsel

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

   utveksling ‒ unikt navn på byttepunktet
   ResponseMatchingTag ‒ lokal etikett for behandling av svaret. For eksempel ved sending av flere identiske forespørsler som tilhører forskjellige brukere.
   RequestDefinition - forespørselsorgan
   Håndteringsprosess ‒ PID for føreren. Denne prosessen vil motta et svar fra serveren.

2. Behandler svaret

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

   ResponsePayload - serverrespons.

For serveren består prosessen også av 2 faser:

1. Initialiserer byttepunktet
2. Behandling av mottatte forespørsler

La oss illustrere denne malen med kode. La oss si at vi må implementere en enkel tjeneste som gir en enkelt eksakt tidsmetode.

Serverkode

La oss definere tjenestens 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{}
}).

La oss definere tjenestekontrolleren 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.

Kundekode

For å sende en forespørsel til tjenesten, kan du kalle opp meldingsforespørsels-API hvor som helst i klienten:

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

I et distribuert system kan konfigurasjonen av komponenter være svært forskjellig, og på tidspunktet for forespørselen kan det hende at meldinger ikke starter ennå, eller tjenestekontrolleren vil ikke være klar til å betjene forespørselen. Derfor må vi sjekke meldingssvaret og håndtere feilsaken.
Etter vellykket sending vil klienten motta et svar eller en feilmelding fra tjenesten.
La oss håndtere begge sakene 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 Response

Det er best å unngå å sende store meldinger. Reaksjonsevnen og stabil drift av hele systemet avhenger av dette. Hvis svaret på en spørring tar opp mye minne, er det obligatorisk å dele det opp i deler.

La meg gi deg et par eksempler på slike tilfeller:

• Komponentene utveksler binære data, for eksempel filer. Å dele opp responsen i små deler hjelper deg med å jobbe effektivt med filer i alle størrelser og unngå minneoverflyt.
• Oppføringer. For eksempel må vi velge alle poster fra en stor tabell i databasen og overføre dem til en annen komponent.

Jeg kaller disse svarene lokomotiv. Uansett er 1024 meldinger på 1 MB bedre enn en enkelt melding på 1 GB.

I Erlang-klyngen får vi en ekstra fordel - å redusere belastningen på utvekslingspunktet og nettverket, siden svar umiddelbart sendes til mottakeren, utenom utvekslingspunktet.

Svar med forespørsel

Dette er en ganske sjelden modifikasjon av RPC-mønsteret for å bygge dialogsystemer.

Publiser-abonner (datadistribusjonstre)

Hendelsesdrevne systemer leverer dem til forbrukere så snart dataene er klare. Dermed er systemer mer utsatt for en push-modell enn for en pull- eller poll-modell. Denne funksjonen lar deg unngå å sløse med ressurser ved å stadig be om og vente på data.
Figuren viser prosessen med å distribuere en melding til forbrukere som abonnerer på et bestemt emne.

Klassiske eksempler på bruk av dette mønsteret er fordelingen av staten: spillverdenen i dataspill, markedsdata på børser, nyttig informasjon i datafeeder.

La oss se 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.

Kilden kan kalle funksjonen for å publisere en melding på et hvilket som helst passende sted:

messaging:publish_message(Exchange, Key, Message).

utveksling - navn på byttepunktet,
nøkkel - rutenøkkel
Melding - nyttelast

Invertert Publiser-abonner

Ved å utvide pub-sub kan du få et mønster som er praktisk for logging. Settet med kilder og forbrukere kan være helt annerledes. Figuren viser en sak med én forbruker og flere kilder.

Oppgavefordelingsmønster

Nesten hvert prosjekt involverer utsatte behandlingsoppgaver, som å generere rapporter, levere varsler og hente data fra tredjepartssystemer. Gjennomstrømningen til systemet som utfører disse oppgavene kan enkelt skaleres ved å legge til behandlere. Alt som gjenstår for oss er å danne en klynge av prosessorer og fordele oppgaver jevnt mellom dem.

La oss se på situasjonene som oppstår ved å bruke eksemplet med 3 behandlere. Selv på oppgavefordelingsstadiet oppstår spørsmålet om rettferdighet i distribusjonen og overløp av behandlere. Round-robin distribusjon vil være ansvarlig for rettferdighet, og for å unngå en situasjon med overløp av behandlere, vil vi innføre en begrensning prefetch_limit. Under forbigående forhold prefetch_limit vil hindre en behandler fra å motta alle oppgaver.

Meldingstjenester administrerer køer og behandlingsprioritet. Prosessorer mottar oppgaver etter hvert som de kommer. Oppgaven kan fullføres eller mislykkes:

• messaging:ack(Tack) - ringes opp hvis meldingen er vellykket behandlet
• messaging:nack(Tack) - ringte inn alle nødsituasjoner. Når oppgaven er returnert, vil meldinger sende den videre til en annen behandler.

Anta at det oppstod en kompleks feil under behandling av tre oppgaver: prosessor 1, etter å ha mottatt oppgaven, krasjet uten å ha tid til å rapportere noe til utvekslingspunktet. I dette tilfellet vil utvekslingspunktet overføre oppgaven til en annen behandler etter at ack timeout er utløpt. Av en eller annen grunn forlot handler 3 oppgaven og sendte nack; som et resultat ble oppgaven også overført til en annen behandler som fullførte den.

Foreløpig oppsummering

Vi har dekket de grunnleggende byggesteinene til distribuerte systemer og fått en grunnleggende forståelse av bruken av dem i Erlang/Elixir.

Ved å kombinere grunnleggende mønstre kan du bygge komplekse paradigmer for å løse nye problemer.

I den siste delen av serien vil vi se på generelle spørsmål om organisering av tjenester, ruting og balansering, og også snakke om den praktiske siden av skalerbarhet og feiltoleranse til systemer.

Slutten av andre del.

Bilde Marius Christensen
Illustrasjoner utarbeidet ved hjelp av websequencediagrams.com

Kilde: www.habr.com