Строителни блокове на разпределени приложения. Първо приближение

В миналото статии Разгледахме теоретичните основи на реактивната архитектура. Време е да поговорим за потоците от данни, пътищата за реализиране на реактивни системи Erlang/Elixir и шаблоните за обмен на съобщения в тях:

• Заявка-отговор
• Заявка-Chunked отговор
• Отговор с заявка
• Публикуване-подписване
• Обърнато публикуване-подписване
• Разпределение на задачи

SOA, MSA и обмен на съобщения

SOA, MSA – системни архитектури, които определят правилата за построяване на системи, докато messaging предоставя примитиви за тяхната реализация.

Не искам да пропагандирам определена архитектура на системи. Подкрепям прилагането на максимално ефективни и полезни практики за конкретен проект и бизнес. Каквато и парадигма да изберем, изграждането на системни блокове е най-добре да се извършва с оглед на Unix начина: компоненти с минимална свързаност, отговарящи за отделни същности. Методите на API извършват максимално прости действия със същностите.

Messaging ‒ както е видно от името ‒ е брокер на съобщения. Основната му цел е да приема и отправя съобщения. Той отговаря за интерфейсите за изпращане на информация, формирането на логически канали за предаване на информация вътре в системата, маршрутизация и балансиране, както и за обработка на откази на системно ниво.
Разработваният messaging не се опитва да конкурира rabbitmq или да го замени. Основните му характеристики са:

• Разпределеност.
  Точки за обмен могат да бъдат създадени на всички възли в кластера, максимално близо до кода, който ги използва.
• Леснота.
  Ориентираност към минимизиране на шаблонния код и удобството на използване.
• Най-добра производителност.
  Не се опитваме да повторим функционалността на rabbitmq, а изтъняваме само архитектурния и транспортния слой, който максимално лесно интегрираме в OTP, минимизирайки разходите.
• Гъвкавост.
  Всеки сервис може да обединява множество шаблони за обмен.
• Отказоустойчивост, внедрена в дизайна.
• Мащабируемост.
  Messaging расте заедно с приложението. С увеличаване на натоварването, точките за обмен могат да бъдат изнесени на отделни машини.

Забележка. От гледна точка на организацията на кода, за сложни системи на Erlang/Elixir мета-проектите са подходящи. Целият код на проекта се намира в едно хранилище - зонтичен проект. При това микросервисите са максимално изолирани и изпълняват прости операции, отговорни за отделна същност. При този подход е лесно да се поддържа API на цялата система, лесно се внасят промени и удобно е да се пишат юнит и интеграционни тестове.

Компонентите на системата взаимодействат директно или чрез брокер. От гледна точка на messaging, всеки сервис има няколко фази на живот:

• Инициализация на сервиса.
  В този етап става конфигурация и стартиране на изпълнителния процес на сервиса и зависимостите.
• Създаване на точка за обмен.
  Сервисът може да използва статична точка за обмен, зададена в конфигурацията на възела, или да създава точки за обмен динамично.
• Регистрация на сервиса.
  За да може сервиса да обслужва заявки, той трябва да бъде регистриран на точката за обмен.
• Нормално функциониране.
  Сервисът извършва полезна работа.
• Завършване на работа.
  Възможни са 2 вида завършване на работа: нормално и аварийно. При нормално обслужването сервиса се изключва от точката за обмен и спира. В аварийни случаи messaging изпълнява един от сценарите за обработка на откази.

Изглежда доста сложно, но в кода не е всичко толкова страшно. Примери с код и коментари ще бъдат предоставени в разобора на шаблони по-късно.

Exchanges

Точката за обмен - процес на messaging, реализиращ логиката на взаимодействие с компонентите в рамките на шаблона за обмен на съобщения. В всички примери, представени по-долу, компонентите взаимодействат чрез точки за обмен, чието съчетание образува messaging.

Шаблони за обмен на съобщения (MEPs)

В глобален мащаб шаблоните за обмен могат да бъдат разделени на двустранни и едностранни. Първите предвиждат отговор на полученото съобщение, вторите не. Класически пример за двустранен шаблон в клиент-серверната архитектура е шаблонът Request-response. Нека разгледаме шаблона и неговите модификации.

Request–response или RPC

RPC се използва, когато ни е нужен отговор от друг процес. Този процес може да бъде стартиран на същия възел или да се намира на друг континент. По-долу е представена схемата на взаимодействие между клиента и сървър чрез messaging.

Тъй като messaging е напълно асинхронен, за клиента обменът се разделя на 2 фази:

1. Изпращане на заявката

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

   Обмен ‒ уникално име на точката за обмен
   ResponseMatchingTag ‒ локален етикет за обработка на отговора. Например, в случай на изпращане на няколко идентични заявки, принадлежащи на различни потребители.
   RequestDefinition ‒ тяло на заявката
   HandlerProcess ‒ PID на обработчика. На този процес ще пристигне отговор от сървъра.

2. Обработка на отговора

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

   ResponsePayload ‒ отговор на сървъра.

За сървъра процесът също се състои от 2 фази:

1. Инициализация на точката за обмен
2. Обработка на получените заявки

Нека илюстрираме този шаблон с код. Да предположим, че е необходимо да реализираме прост сервис, предоставящ единствен метод за точно време.

Код на сървъра

Нека извлечем определението на API на сервиза в 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{}
}).

Да определим контролера на сервиза в 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.

Код на клиента

За да изпратите заявка до сервиза, можете да извикате messaging request API от всяка точка на клиента:

case messaging:request(?EXCHANGE, tag, #time_req{opts = #{}}, self()) of
    ok -> ok;
    _ -> %% логика за повторение или неуспех
end

В разпределената система конфигурацията на компонентите може да бъде много различна и в момента на заявката messaging може да не е стартиран, или контролерът на сервиза да не е готов да обслужи искането. Затова е необходимо да проверим отговора на messaging и да обработим случая на отказ.
След успешно изпращане, клиентът ще получи отговор или грешка от сервиза.
Да обработим и двата случая в 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};

Заявка-Chunked отговор

Необходимо да се избягва предаването на огромни съобщения. Тази характеристика оказва влияние на отзивчивостта и стабилността на цялата система. Ако отговорът на заявката заема много памет, разделяне на части е задължително.

Ще дам няколко примера за такива случаи:

• Компонентите обменят бинарни данни, например файлове. Разделянето на отговора на по-малки части помага за ефективната работа с файлове с всякакъв размер и предотвратява препълване на паметта.
• Листинги. Например, трябва да изберем всички записи от огромна таблица в базата данни и да ги предадем на друг компонент.

Наричам такива отговори "локомотиви". Във всеки случай 1024 съобщения по 1 Мб са по-добре от едно единствено съобщение с размер 1 Гб.

В Erlang клъстера получаваме допълнителна полза - намаляване на натоварването на точката на обмена и мрежата, тъй като отговорите веднага отиват при получателя, заобикаляйки точката на обмена.

Отговор с заявка

Това е доста рядка модификация на шаблона RPC за изграждане на диалогови системи.

Publish-subscribe (дърво за разпределение на данни)

Системи, ориентирани към събития, доставят данните на потребителите при готовност. По този начин системите са по-склонни към push модел, отколкото към pull или poll. Тази особеност позволява да не се изразходват ресурси ненужно, постоянно запитвайки и очаквайки данни.
На схемата е представен процесът на разпространение на съобщение към потребителите, подписали се на определена тема.

Класическите примери за използване на този шаблон включват разпространение на състояние: игровия свят в компютърни игри, пазарни данни на борсите, полезна информация в датафийдове.

Нека разгледаме кода на абоната:

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.

Източникът може да извиква функцията за публикуване на съобщение навсякъде, където е удобно:

messaging:publish_message(Exchange, Key, Message).

Обмен ‒ името на точката за обмен,
Ключ ‒ ключ за маршрутизация
Съобщение ‒ полезен товар

Обърнато публикуване-подписване

След разгръщането на pub-sub, можем да получим шаблон, удобен за регистриране. Наборът от източници и потребители може да бъде напълно различен. На изображението е показан случай с един потребител и множество източници.

Шаблон за разпределение на задачи

В почти всеки проект възникват задачи за отложена обработка, като генериране на отчети, доставка на известия, получаване на данни от външни системи. Пропускливостта на системата, която извършва тези задачи, лесно може да бъде мащабирана чрез добавяне на обработващи единици. Всичко, което ни остава, е да формираме клъстер от обработващи единици и равномерно да разпределяме задачите между тях.

Нека разгледаме възникващите ситуации на примера на 3 обработващи единици. Още на етапа на разпределение на задачите се появява въпросът за справедливостта на разпределението и пренатоварването на обработващите единици. За справедливостта ще отговаря разпределението round-robin, а за да избегнем пренатоварване на обработващите единици, ще въведем ограничение prefetch_limit. В преходни режими prefetch_limit не позволява на една обработваща единица да получи всичките задачи.

Messaging управлява опашките и приоритета на обработката. Обработващите единици получават задачи при тяхното постъпване. Изпълнението на задачата може да завърши успешно или с отказ:

• messaging:ack(Tack) ‒ извиква се в случай на успешно обработване на съобщението
• messaging:nack(Tack) ‒ извиква се при всички извънредни ситуации. След връщане на задачата messaging ще я предаде на друга обработваща единица.

Да предположим, че при обработка на три задачи настъпи сложен отказ: обработваща единица 1 след получаване на задачата е спаднала, без да успее да съобщи нещо на точката за обмен. В този случай точката за обмен след изтичането на ack timeout ще предаде задачата на друга обработваща единица. Обработваща единица 3 по някаква причина е отказала задачата и е изпратила nack, в резултат на което задачата също е преминала на друга обработваща единица, която я е изпълнила успешно.

Предварителен резултат

Изучихме основните строителни елементи на разпределените системи и получихме базово разбиране за тяхното приложение в Erlang/Elixir.

Комбинирайки основните шаблони, можем да изградим сложни парадигми за решаване на възникващите задачи.

В заключителната част на цикъла ще разгледаме общите въпроси за организацията на услугите, маршрутизацията и балансирането, както и ще поговорим за практическата страна на мащабируемостта и устойчивостта на системите.

Край на втората част.

Снимка Мариус Кристенсен
Илюстрациите са подготвени с помощта на websequencediagrams.com

Източник: habr.com