🥇Fundamenty aplikacji rozproszonych. Pierwsze przybliżenie

W poprzedniej artykuł Omówiliśmy teoretyczne podstawy architektury reaktywnej. Czas porozmawiać o strumieniach danych, sposobach realizacji systemów reaktywnych Erlang/Elixir oraz wzorcach wymiany wiadomości w nich:

Żądanie-odpowiedź
Żądanie-odpowiedź dzielona
Odpowiedź z żądaniem
Pub-sub (publikowanie-subskrybowanie)
Odwrócone Pub-sub
Rozdzielanie zadań

SOA, MSA i wymiana wiadomości

SOA, MSA to architektury systemowe, które określają zasady budowania systemów, podczas gdy messaging dostarcza prymitywy do ich realizacji.

Nie chcę propagować żadnej architektury budowy systemów. Opowiadam się za stosowaniem maksymalnie efektywnych i użytecznych praktyk dla konkretnego projektu i biznesu. Niezależnie od wybranej paradygmy, lepiej jest tworzyć bloki systemowe z uwzględnieniem podejścia Unix-way: komponenty o minimalnym powiązaniu, odpowiedzialne za pojedyncze encje. Metody API realizują maksymalnie proste działania na encjach.

Messaging ‒ jak to wynika z nazwy ‒ to broker wiadomości. Jego głównym celem jest przyjmowanie i wysyłanie wiadomości. Odpowiada za interfejsy wysyłania informacji, tworzenie logicznych kanałów przesyłania informacji wewnątrz systemu, routowanie i balansowanie obciążenia, a także przetwarzanie awarii na poziomie systemowym.
Tworzony messaging nie próbuje konkurować z rabbitmq ani go zastępować. Jego główne cechy:

Rozproszenie.
Punkty wymiany można tworzyć na wszystkich węzłach klastra, maksymalnie blisko do kodu, który je wykorzystuje.
Prostota.
Skupienie na minimalizacji kodu szablonowego oraz łatwości użycia.
Najlepsza wydajność.
Nie próbujemy powtórzyć funkcjonalności rabbitmq, a jedynie oddzielamy warstwę architektoniczną i transportową, którą łatwo integrować z OTP, minimalizując koszty.
Elastyczność.
Każda usługa może łączyć w sobie wiele wzorców wymiany.
Odporność na błędy zakorzeniona w projekcie.
Skalowalność.
Messaging rozwija się wraz z aplikacją. W miarę wzrostu obciążenia można przenieść punkty wymiany na oddzielne maszyny.

Uwaga. Z perspektywy organizacji kodu, meta-projekty są doskonałym rozwiązaniem dla złożonych systemów na Erlangu/Elixiru. Cały kod projektu znajduje się w jednym repozytorium ‒ projekcie nadrzędnym. Mikrousługi są maksymalnie izolowane i wykonują proste operacje, odpowiadające za poszczególne jednostki. Takie podejście ułatwia zarządzanie API całego systemu, wprowadzanie zmian oraz pisanie testów jednostkowych i integracyjnych.

Komponenty systemu komunikują się bezpośrednio lub za pośrednictwem brokera. Z perspektywy komunikacji, każda usługa ma kilka faz życia:

Inicjalizacja usługi.
Na tym etapie odbywa się konfiguracja i uruchomienie procesu usługowego oraz zależności.
Tworzenie punktu wymiany.
Usługa może korzystać z statycznego punktu wymiany, określonego w konfiguracji węzła, lub tworzyć punkty wymiany dynamicznie.
Rejestracja usługi.
Aby usługa mogła obsługiwać żądania, należy ją zarejestrować w punkcie wymiany.
Normalne funkcjonowanie.
Usługa wykonuje pożyteczną pracę.
Zakończenie pracy.
Są dwa rodzaje zakończenia pracy: normalne i awaryjne. W przypadku normalnego, usługa jest odłączana od punktu wymiany i zatrzymywana. W sytuacjach awaryjnych komunikacja realizuje jeden z scenariuszy obsługi awarii.

Wygląda na dość skomplikowane, ale w kodzie nie jest tak strasznie. Przykłady kodu z komentarzami będą przedstawione w analizie szablonów nieco później.

Wymiany

Punkt wymiany ‒ proces komunikacji, realizujący logikę interakcji z komponentami w ramach szablonu wymiany wiadomości. We wszystkich przykładach poniżej komponenty komunikują się poprzez punkty wymiany, których kombinacja tworzy komunikację.

Wzory wymiany wiadomości (MEP)

Globalnie, wzory wymiany można podzielić na dwustronne i jednostronne. Pierwsze zakładają odpowiedź na przychodzącą wiadomość, drugie tego nie robią. Klasycznym przykładem dwustronnego wzoru w architekturze klient-serwer jest wzór Request-response. Przyjrzyjmy się wzorowi oraz jego modyfikacjom.

Request–response lub RPC

RPC jest używane, gdy potrzebujemy uzyskać odpowiedź od innego procesu. Ten proces może być uruchomiony na tym samym węźle lub znajdować się na innym kontynencie. Poniżej przedstawiona jest schemat interakcji klienta i serwera poprzez komunikację.

Ponieważ komunikacja jest całkowicie asynchroniczna, wymiana dla klienta dzieli się na 2 fazy:

Wysyłanie żądania
```
messaging:request(Exchange, ResponseMatchingTag, RequestDefinition, HandlerProcess).
```
Exchange ‒ unikalna nazwa punktu wymiany
ResponseMatchingTag ‒ lokalna etykieta do przetwarzania odpowiedzi. Na przykład w przypadku wysyłania wielu identycznych żądań od różnych użytkowników.
RequestDefinition ‒ treść żądania
HandlerProcess ‒ PID przetwarzacza. Ten proces otrzyma odpowiedź od serwera.

Przetwarzanie odpowiedzi

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

ResponsePayload ‒ odpowiedź serwera.

Dla serwera proces składa się z 2 faz:

Inicjalizacja punktu wymiany
Przetwarzanie przychodzących żądań

Ilustrujemy ten szablon kodem. Załóżmy, że musimy zrealizować prostą usługę, która udostępnia jedną metodę zwracającą dokładny czas.

Kod serwera

Wyodrębnijmy definicję API usługi do api.hrl:

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

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

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

Zdefiniujmy kontroler usługi w time_controller.erl

%% W przykładzie pokazany jest tylko wartościowy kod. Wstawiając go do szablonu gen_server, można uzyskać działającą usługę.

%% inicjalizacja gen_server
init(Args) ->
  %% połączenie z punktem wymiany
  messaging:monitor_exchange(req_resp, ?EXCHANGE, default, self())
  {ok, #{}}.

%% obsługa zdarzenia utraty połączenia z punktem wymiany. To samo zdarzenie pojawia się, jeśli punkt wymiany jeszcze nie został uruchomiony.
handle_info(#exchange_die{exchange = ?EXCHANGE}, State) ->
  erlang:send(self(), monitor_exchange),
  {noreply, State};

%% obsługa 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};

%% zakończenie pracy gen_server
terminate(_Reason, _State) ->
  messaging:demonitor_exchange(req_resp, ?EXCHANGE, default, self()),
  ok.

Kod klienta

Aby wysłać żądanie do usługi, w dowolnym miejscu klienta można wywołać interfejs API messaging request:

case messaging:request(?EXCHANGE, tag, #time_req{opts = #{}}, self()) of
    ok -> ok;
    _ -> %% logika powtórzenia lub błędu
end

W rozproszonym systemie konfiguracja komponentów może być bardzo różna, a w momencie żądania messaging może jeszcze nie być uruchomiony, albo kontroler usługi nie jest gotowy, aby obsłużyć żądanie. Dlatego musimy sprawdzić odpowiedź messaging i obsłużyć przypadek odmowy.
Po pomyślnym wysłaniu klient otrzyma odpowiedź lub błąd od usługi.
Obsłużmy oba przypadki w 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};

Żądanie-odpowiedź dzielona

Lepiej unikać przesyłania dużych wiadomości. Zależy od tego responsywność i stabilność całego systemu. Jeśli odpowiedź na zapytanie zajmuje dużo pamięci, podział na części jest obowiązkowy.

Podam kilka przykładów takich przypadków:

Komponenty wymieniają się danymi binarnymi, na przykład plikami. Podział odpowiedzi na małe części umożliwia efektywne przetwarzanie plików o dowolnym rozmiarze i unikanie przepełnienia pamięci.
Listy. Na przykład, musimy wybrać wszystkie zapisy z ogromnej tabeli w bazie danych i przekazać je innemu komponentowi.

Nazywam takie odpowiedzi lokomotywą. W każdym razie, 1024 wiadomości po 1 MB jest lepsze niż jedna wiadomość o wielkości 1 GB.

W klastrze Erlang uzyskujemy dodatkowy zysk ‒ zmniejszenie obciążenia punktu wymiany i sieci, ponieważ odpowiedzi są od razu kierowane do odbiorcy, omijając punkt wymiany.

Odpowiedź z żądaniem

Jest to dość rzadkie modyfikowanie wzorca RPC do budowania systemów dialogowych.

Systemy zorientowane na zdarzenia dostarczają dane konsumentom w miarę ich gotowości. W ten sposób systemy są bardziej skłonne do modelu push niż pull czy poll. Ta cecha pozwala nie marnować zasobów na ciągłe pytanie i oczekiwanie na dane.
Na ilustracji przedstawiono proces rozprzestrzeniania wiadomości do konsumentów, którzy są subskrybowani na dany temat.

Klasycznymi przykładami użycia tego wzorca jest dystrybucja stanu: świata gry w grach komputerowych, danych rynkowych na giełdach, użytecznych informacji w zasilaczach danych.

Rozważmy kod subskrybenta:

init(_Args) ->
  %% subskrybujemy się na punkt wymiany, klucz = key
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {ok, #{}}.

handle_info(#exchange_die{exchange = ?SUBSCRIPTION}, State) ->
  %% jeśli punkt wymiany jest niedostępny, próbujemy ponownie się połączyć
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {noreply, State};

%% przetwarzamy nadchodzące wiadomości
handle_info(#'$msg'{exchange = ?SUBSCRIPTION, message = Msg}, State) ->
  ?debugVal(Msg),
  {noreply, State};

%% przy zatrzymaniu konsumenta - rozłączamy się od punktu wymiany
terminate(_Reason, _State) ->
  messaging:unsubscribe(?SUBSCRIPTION, key, tag, self()),
  ok.

Źródło może wywołać funkcję publikacji wiadomości w dowolnym miejscu:

messaging:publish_message(Exchange, Key, Message).

Exchange ‒ nazwa punktu wymiany,
Klucz ‒ klucz routingu
Wiadomość ‒ ładunek użyteczny

Rozwijając pub-sub, można uzyskać wzór odpowiedni do logowania. Zestaw źródeł i konsumentów może być zupełnie różny. Na rysunku przedstawiono przypadek z jednym konsumentem i wieloma źródłami.

Wzór dystrybucji zadań

W prawie każdym projekcie występują zadania o odroczonym przetwarzaniu, takie jak generowanie raportów, dostarczanie powiadomień, uzyskiwanie danych z zewnętrznych systemów. Przepustowość systemu wykonującego te zadania łatwo skalować poprzez dodanie przetwórców. Wszystko, co nam pozostaje, to utworzyć klaster przetwórców i równomiernie rozdzielać zadania między nimi.

Rozważmy sytuacje, które mogą wystąpić na przykładzie trzech przetwórców. Już na etapie rozdzielania zadań pojawia się kwestia sprawiedliwości dystrybucji oraz przepełnienia przetwórców. Za sprawiedliwość odpowiada dystrybucja round-robin, a aby uniknąć sytuacji przepełnienia przetwórców, wprowadzimy ograniczenie prefetch_limit. W trybach przejściowych prefetch_limit nie pozwoli jednemu przetwórcy na otrzymanie wszystkich zadań.

Messaging zarządza kolejkami i priorytetami przetwarzania. Przetwórcy otrzymują zadania w miarę ich przybywania. Wykonanie zadania może zakończyć się sukcesem lub niepowodzeniem:

messaging:ack(Tack) ‒ wywoływane w przypadku pomyślnego przetworzenia wiadomości
messaging:nack(Tack) ‒ wywoływane w nieprzewidzianych sytuacjach. Po zwróceniu zadania messaging przekaże je innemu przetwórcy.

Przypuśćmy, że podczas przetwarzania trzech zadań wystąpiła poważna awaria: przetwórca 1 po otrzymaniu zadania upadł, nie zdążywszy przekazać nic punktowi wymiany. W takim przypadku punkt wymiany, po upływie czasu na ack, przekaże zadanie innemu przetwórcy. Przetwórca 3 z jakiegoś powodu odmówił przyjęcia zadania i wysłał nack, w efekcie zadanie również trafiło do innego przetwórcy, który je pomyślnie wykonał.

Wstępny podsumowanie

Omówiliśmy podstawowe elementy systemów rozproszonych i uzyskaliśmy podstawowe zrozumienie ich zastosowania w Erlang'u/Elixirze.

Łącząc podstawowe wzory, można budować złożone paradygmaty do rozwiązywania pojawiających się zadań.

W ostatniej części cyklu omówimy ogólne zagadnienia dotyczące organizacji usług, routingu i balansowania, a także poruszymy praktyczne aspekty skalowalności i odporności systemów na awarie.

Koniec drugiej części.

Zdjęcie Marius Christensen
Ilustracje przygotowano za pomocą websequencediagrams.com

Źródło: habr.com