Будівельні блоки розподілених програм. Перше наближення

У минулій статті ми розібрали теоретичні засади реактивної архітектури. Настав час поговорити про потоки даних, шляхи реалізації реактивних Erlang/Elixir систем та шаблони обміну повідомленнями в них:

• Request-response
• Request-Chunked Response
• Response with Request
• Опублікувати-підписатися
• Inverted Publish-subscribe
• Task distribution

SOA, MSA та обмін повідомленнями

SOA, MSA – системні архітектури, що визначають правила побудови систем, у той час як messaging надає примітиви для їх реалізації.

Я не хочу пропагувати ту чи іншу архітектуру побудови систем. Я за застосування максимально ефективних та корисних для конкретного проекту та бізнесу практик. Яку б парадигму ми не вибрали, створювати системні блоки краще з огляду на Unix-way: компоненти з мінімальним зв'язком, які відповідають за окремі сутності. Методи API виконують максимально прості дії із сутностями.

Messaging ‒ як відомо з назви ‒ брокер повідомлень. Його основна мета – приймати та віддавати повідомлення. Він відповідає за інтерфейси надсилання інформації, формування логічних каналів передачі інформації всередині системи, маршрутизацію та балансування, а також обробку відмов на системному рівні.
Розроблений messaging не намагається конкурувати з rabbitmq або замінювати його. Його основні риси:

• Розподіл.
  Точки обміну можна створювати на всіх вузлах кластера, максимально близько до коду, який їх використовує.
• Простота.
  Спрямованість на мінімізацію шаблонного коду та зручність використання.
• Висока якість.
  Ми не намагаємося повторити функціонал rabbitmq, а виділяємо лише архітектурний та транспортний шар, який максимально просто вписуємо в OTP, мінімізуючи витрати.
• Гнучкість.
  Кожен сервіс може об'єднувати безліч шаблонів обміну.
• Відмовостійкість, закладена у дизайні.
• Масштабованість.
  Messaging росте разом із додатком. У міру збільшення навантаження можна виносити точки обміну окремі машини.

Зауваження. З точки зору організації коду, для складних систем Erlang/Elixir добре підходять мета-проекти. Весь код проекту знаходиться в одному репозиторії – парасольковому проекті. При цьому мікросервіси максимально ізольовані та виконують прості операції, що відповідають за окрему сутність. За такого підходу легко підтримувати API всієї системи, просто вносити зміни, зручно писати юніт та інтеграційні тести.

Компоненти системи взаємодіють безпосередньо або через брокера. З позиції messaging, кожен сервіс має кілька життєвих фаз:

• Ініціалізація сервісу.
  На даному етапі відбувається конфігурація та запуск виконуючого обслуговування процесу та залежностей.
• Створення точки обміну.
  Сервіс може використовувати статичну точку обміну, задану в конфігурації вузла, або створювати точки обміну динамічно.
• Реєстрація сервісу.
  Щоб сервіс міг обслуговувати запити, його потрібно зареєструвати на точці обміну.
• Нормальне функціонування.
  Сервіс виконує корисну роботу.
• Завершення роботи.
  Можливі 2 види завершення роботи: штатне та аварійне. При штатному сервіс відключається від точки обміну та зупиняється. В аварійних випадках messaging виконує один із сценаріїв обробки відмов.

Виглядає досить складно, але в коді не так страшно. Приклади коду з коментарями будуть наведені в аналізі шаблонів трохи пізніше.

Біржі

Точка обміну ‒ процес messaging, що реалізує логіку взаємодії з компонентами у рамках шаблону обміну повідомленнями. У всіх прикладах, наведених нижче, компоненти взаємодіють через точки обміну, комбінація яких і утворює messaging.

Message exchange patterns (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;
    _ -> %% repeat or fail logic
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};

Request-Chunked Response

Краще не допускати передачі величезних повідомлень. Від цього залежить чуйність та стабільна робота всієї системи. Якщо відповідь на запит займає багато пам'яті, розбивка на частини є обов'язковою.

Наведу кілька прикладів таких випадків:

• Компоненти обмінюються бінарними даними, наприклад, файлами. Розбивка відповіді на невеликі частини допомагає ефективно працювати з файлами будь-якого розміру та не ловити переповнення пам'яті.
• Лістинги. Наприклад, потрібно вибрати всі записи з величезної таблиці в базі і передати іншому компоненту.

Я називаю такі відповіді паровозом. У будь-якому випадку, 1024 повідомлень по 1 Мб краще, ніж єдине повідомлення розміром 1 Гб.

У кластері Erlang ми отримуємо додатковий виграш – зменшення навантаження на точку обміну та мережу, оскільки відповіді відразу направляються одержувачу, минаючи точку обміну.

Response with Request

Це досить рідкісна модифікація патерну RPC для побудови діалогових систем.

Publish-subscribe (data distribution tree)

Подієво-орієнтовані системи в міру готовності даних доставляють їх споживачам. Таким чином, системи більш схильні до 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).

обмін ‒ назва точки обміну,
ключ ‒ ключ маршрутизації
Повідомлення - корисне навантаження

Inverted Publish-subscribe

Розгорнувши pub-sub, можна отримати патерн, зручний для логування. Набір джерел та споживачів може бути зовсім різним. На малюнку представлений випадок з одним споживачем та безліччю джерел.

Task distribution pattern

Майже у кожному проекті виникають завдання відкладеної обробки, такі як формування звітів, доставка повідомлень, отримання даних із сторонніх систем. Пропускна здатність системи, яка виконує ці завдання, легко масштабується шляхом додавання обробників. Все, що нам залишається – це сформувати кластер з обробників і рівномірно розподіляти завдання між ними.

Розглянемо ситуації, що виникають на прикладі 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