Stavební bloky distribuovaných aplikací. První přístup

V minulosti článek Prozkoumali jsme teoretické základy reaktivní architektury. Je čas mluvit o datových tocích, způsobech implementace reaktivních systémů Erlang/Elixir a vzorcích zasílání zpráv v nich:

• Vyžádat odpověď
• Request-Chunked Response
• Odpovězte žádostí
• Publikovat-předplatit
• Invertovaný Publikovat-předplatit
• Rozdělení úkolů

SOA, MSA a zasílání zpráv

SOA, MSA jsou systémové architektury, které definují pravidla pro budování systémů, zatímco zasílání zpráv poskytuje primitiva pro jejich implementaci.

Nechci propagovat tu či onu systémovou architekturu. Jsem pro používání nejúčinnějších a nejužitečnějších postupů pro konkrétní projekt a podnikání. Ať už zvolíme jakékoli paradigma, je lepší vytvářet systémové bloky s ohledem na unixovou cestu: komponenty s minimální konektivitou, zodpovědné za jednotlivé entity. Metody API provádějí s entitami nejjednodušší možné akce.

Messaging je, jak název napovídá, zprostředkovatel zpráv. Jeho hlavním účelem je přijímat a odesílat zprávy. Zodpovídá za rozhraní pro odesílání informací, vytváření logických kanálů pro přenos informací v rámci systému, směrování a vyvažování, jakož i zpracování chyb na úrovni systému.
Zasílání zpráv, které vyvíjíme, se nesnaží konkurovat nebo nahradit rabbitmq. Jeho hlavní rysy:

• Rozdělení.
  Výměnné body lze vytvořit na všech uzlech clusteru, co nejblíže kódu, který je používá.
• Jednoduchost.
  Zaměřte se na minimalizaci standardního kódu a snadné použití.
• Lepší výkon.
  Nesnažíme se zopakovat funkcionalitu rabbitmq, ale vyzdvihnout pouze architektonickou a transportní vrstvu, kterou co nejjednodušeji vložíme do OTP, minimalizujeme náklady.
• Flexibilita.
  Každá služba může kombinovat mnoho výměnných šablon.
• Odolnost podle návrhu.
• Škálovatelnost.
  Zasílání zpráv roste s aplikací. Se zvyšujícím se zatížením můžete body výměny přesouvat na jednotlivé stroje.

Poznámka. Z hlediska organizace kódu jsou metaprojekty vhodné pro komplexní systémy Erlang/Elixir. Veškerý kód projektu je umístěn v jednom úložišti – zastřešujícím projektu. Mikroslužby jsou přitom maximálně izolované a provádějí jednoduché operace, které má na starosti samostatný subjekt. S tímto přístupem je snadné udržovat API celého systému, je snadné provádět změny, je vhodné psát unit a integrační testy.

Systémové komponenty interagují přímo nebo prostřednictvím zprostředkovatele. Z hlediska zasílání zpráv má každá služba několik životních fází:

• Inicializace služby.
  V této fázi je nakonfigurován a spuštěn proces spouštějící službu a její závislosti.
• Vytvoření výměnného bodu.
  Služba může používat statický výměnný bod zadaný v konfiguraci uzlu nebo vytvářet výměnné body dynamicky.
• Registrace služby.
  Aby služba obsluhovala požadavky, musí být zaregistrována na výměnném místě.
• Normální operace.
  Služba vytváří užitečnou práci.
• Vypnout.
  Jsou možné 2 typy vypnutí: normální a nouzové. Při běžném provozu je služba odpojena od výměnného místa a zastaví se. V nouzových situacích zasílání zpráv spustí jeden ze skriptů pro převzetí služeb při selhání.

Vypadá to docela složitě, ale kód není až tak děsivý. Příklady kódu s komentáři budou uvedeny v analýze šablon o něco později.

Směnárny

Exchange point je proces zasílání zpráv, který implementuje logiku interakce s komponentami v šabloně zasílání zpráv. Ve všech níže uvedených příkladech komponenty interagují prostřednictvím výměnných bodů, jejichž kombinace tvoří zasílání zpráv.

Vzory výměny zpráv (poslanci)

Globálně lze směnné vzory rozdělit na obousměrné a jednosměrné. První znamená odpověď na příchozí zprávu, druhá nikoli. Klasickým příkladem obousměrného vzoru v architektuře klient-server je vzor požadavek-odpověď. Podívejme se na šablonu a její úpravy.

Požadavek-odpověď nebo RPC

RPC se používá, když potřebujeme obdržet odpověď od jiného procesu. Tento proces může být spuštěn na stejném uzlu nebo umístěn na jiném kontinentu. Níže je schéma interakce mezi klientem a serverem prostřednictvím zpráv.

Vzhledem k tomu, že zasílání zpráv je zcela asynchronní, je výměna pro klienta rozdělena do 2 fází:

1. Odeslání žádosti

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

   výměna ‒ jedinečný název výměnného místa
   ResponseMatchingTag ‒ místní štítek pro zpracování odpovědi. Například v případě odeslání několika stejných požadavků, které patří různým uživatelům.
   Požadavek Definice - tělo žádosti
   HandlerProcess ‒ PID handlera. Tento proces obdrží odpověď ze serveru.

2. Zpracování odpovědi

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

   ResponsePayload - odpověď serveru.

U serveru se proces také skládá ze 2 fází:

1. Inicializace výměnného bodu
2. Zpracování přijatých požadavků

Pojďme si tuto šablonu ilustrovat kódem. Řekněme, že potřebujeme implementovat jednoduchou službu, která poskytuje jedinou metodu přesného času.

Kód serveru

Pojďme definovat API služby v 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{}
}).

Definujme řadič služby v 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.

Kód klienta

Chcete-li odeslat požadavek službě, můžete zavolat rozhraní API žádosti o zasílání zpráv kdekoli v klientovi:

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

V distribuovaném systému může být konfigurace komponent velmi odlišná a v době požadavku se ještě nemusí zasílání zpráv spustit nebo servisní kontrolér nebude připraven požadavek obsloužit. Proto musíme zkontrolovat odezvu na zasílání zpráv a vyřešit případ selhání.
Po úspěšném odeslání klient obdrží odpověď nebo chybu ze služby.
Vyřešme oba případy v 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

Nejlepší je vyhnout se odesílání velkých zpráv. Na tom závisí odezva a stabilní provoz celého systému. Pokud odpověď na dotaz zabírá hodně paměti, pak je její rozdělení na části povinné.

Dovolte mi uvést několik příkladů takových případů:

• Komponenty si vyměňují binární data, jako jsou soubory. Rozdělení odezvy na malé části vám pomůže efektivně pracovat se soubory jakékoli velikosti a vyhnout se přetečení paměti.
• Výpisy. Potřebujeme například vybrat všechny záznamy z obrovské tabulky v databázi a přenést je do jiné komponenty.

Těmto odpovědím říkám lokomotiva. V každém případě je 1024 zpráv o velikosti 1 MB lepších než jedna zpráva o velikosti 1 GB.

V clusteru Erlang získáváme další výhodu – snížení zátěže výměnného bodu a sítě, protože odpovědi jsou okamžitě odesílány příjemci a obcházejí směnný bod.

Odpovězte žádostí

Toto je poměrně vzácná modifikace vzoru RPC pro vytváření dialogových systémů.

Publish-subscribe (strom distribuce dat)

Systémy řízené událostmi je doručují spotřebitelům, jakmile jsou data připravena. Systémy jsou tedy náchylnější k modelu push než modelu pull nebo poll. Tato funkce vám umožňuje vyhnout se plýtvání zdroji neustálým vyžadováním dat a čekáním na ně.
Obrázek ukazuje proces distribuce zprávy spotřebitelům přihlášeným k odběru konkrétního tématu.

Klasickými příklady použití tohoto vzoru jsou rozložení stavu: herní svět v počítačových hrách, tržní data o burzách, užitečné informace v datových zdrojích.

Podívejme se na kód předplatitele:

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.

Zdroj může zavolat funkci a publikovat zprávu na libovolném vhodném místě:

messaging:publish_message(Exchange, Key, Message).

výměna - název směnárny,
Klíč - směrovací klíč
Zpráva - užitečné zatížení

Invertovaný Publikovat-předplatit

Rozšířením pub-sub můžete získat vzor vhodný pro protokolování. Soubor zdrojů a spotřebitelů může být zcela odlišný. Obrázek ukazuje případ s jedním spotřebitelem a více zdroji.

Vzor rozdělení úkolů

Téměř každý projekt zahrnuje odložené úlohy zpracování, jako je generování zpráv, doručování oznámení a získávání dat ze systémů třetích stran. Propustnost systému provádějícího tyto úkoly lze snadno škálovat přidáním manipulátorů. Zbývá nám jen vytvořit shluk procesorů a rovnoměrně mezi ně rozdělit úkoly.

Podívejme se na vzniklé situace na příkladu 3 handlerů. Již ve fázi rozdělování úkolů vyvstává otázka spravedlnosti rozdělování a přetečení handlerů. Za férovost bude zodpovědná kruhová distribuce, a abychom předešli situaci přetečení handlerů, zavedeme omezení prefetch_limit. V přechodných podmínkách prefetch_limit zabrání jednomu psovodovi přijímat všechny úkoly.

Zasílání zpráv spravuje fronty a prioritu zpracování. Zpracovatelé dostávají úkoly tak, jak přicházejí. Úloha může být dokončena úspěšně nebo selže:

• messaging:ack(Tack) - volá se, pokud je zpráva úspěšně zpracována
• messaging:nack(Tack) - volán ve všech nouzových situacích. Jakmile je úloha vrácena, zasílání zpráv ji předá jinému handleru.

Předpokládejme, že při zpracování tří úloh došlo ke komplexnímu selhání: procesor 1 po obdržení úlohy spadl, aniž by měl čas cokoli nahlásit do výměnného bodu. V tomto případě výměnný bod převede úlohu na jiného handlera po vypršení časového limitu pro potvrzení. Z nějakého důvodu handler 3 opustil úkol a poslal nack, v důsledku toho byl úkol také převeden na jiného handlera, který jej úspěšně dokončil.

Předběžné shrnutí

Pokryli jsme základní stavební bloky distribuovaných systémů a získali základní znalosti o jejich použití v Erlang/Elixir.

Kombinací základních vzorů můžete vytvářet komplexní paradigmata k řešení vznikajících problémů.

V závěrečném díle seriálu se podíváme na obecné otázky organizace služeb, směrování a vyvažování a také si povíme o praktické stránce škálovatelnosti a odolnosti systémů proti chybám.

Konec druhého dílu.

Fotografie Marius Christensen
Ilustrace připravené pomocí websequencediagrams.com

Zdroj: www.habr.com