Это перевод статьи Руководство по событийно-ориентированной архитектуре в Elixir, написанной Сапаном Дивакаром.

Я также добавил несколько изображений, чтобы лучше проиллюстрировать объяснение. Если вы сомневаетесь в переводе, посетите оригинальный пост. Давайте перейдем к посту.

В этом посте мы рассмотрим, как событийно-ориентированная архитектура может сделать ваше приложение более отзывчивым для пользователей и развязать ваши модули для лучшего опыта разработчика.

Мы также рассмотрим различные методы реализации событийно-управляемой архитектуры с помощью Elixir. Язык Elixir особенно хорошо подходит для этого, поскольку он предлагает продвинутые и лаконичные API передачи сообщений и отличную поддержку параллелизма в BEAM.

Но сначала: что именно представляет собой событийно-управляемая архитектура?

• Архитектура, управляемая событиями: Введение
• Строительные блоки архитектуры, управляемой событиями
• Преимущества событийно-ориентированной архитектуры
• Подходы
  • Синхронная событийно-управляемая архитектура в Elixir
  • Использование GenServer в событийно-управляемой архитектуре в Elixir
  • Событийно-управляемая реализация с использованием GenStage на Elixir
• Заключение: Событийно-управляемая архитектура в Elixir – выход за рамки GenStage

Событийно-управляемая архитектура: Введение

Событийно-ориентированная архитектура – это архитектура, в которой события управляют поведением и потоком вашего приложения. Основными компонентами архитектуры являются производители событий (event producers), шина событий (event bus) и потребители событий (event consumers).

Событием может быть все, что представляет собой изменение состояния системы. Например, в приложении электронной коммерции “e-commerce” покупка продукта пользователем может вызвать событие продажи, после чего потребитель может инициировать процесс обновления запасов.

Архитектура, основанная на событиях, позволяет приложениям действовать в соответствии с событиями по мере их возникновения. Различные части приложения функционируют и развиваются относительно независимо друг от друга в хорошо продуманном, основанном на событиях дизайне. Организации могут назначать отдельные команды для работы с отдельными частями приложения и оптимизировать рабочий процесс. Это также создает четкую границу между различными частями приложения, помогая в будущих упражнениях по масштабированию.

Событийно-управляемая архитектура приобрела популярность прежде всего в продуктах на базе микросервисов, но ее можно использовать и для монолита.

Все всегда становится понятнее на примере, поэтому давайте рассмотрим один из них.

Строительные блоки архитектуры, управляемой событиями

Давайте подробно обсудим каждый строительный блок на примере приложения “электронной коммерции”.

Представьте, что пользователь совершает новую покупку на сайте. Новый заказ – это событие, генерируемое частью, которая управляет заказом: производителем событий.

Событие может быть передано на шину событий. Шина событий может быть любой, например:

• Таблица в базе данных.
• Очередь событий в памяти приложения.
• Внешний инструмент, такой как RabbitMQ или Apache Kafka.

Производители мероприятий, заинтересованные в проведении такого рода мероприятий, могут записаться на автобус для проведения мероприятий. Событие передается им, и они выполняют некоторую обработку поверх него.

Например, система управления запасами подпишется (subscribe) на событие нового заказа и обновит запасы продукции. Другая система также может выбрать то же событие в заказе – например, служба выполнения заказов может обработать это событие и создать маршрут доставки продукта.

Преимущества событийно-ориентированной архитектуры

Есть несколько преимуществ использования архитектуры, управляемой событиями, а не той, которая обрабатывает все последовательно.

Событийно-ориентированная архитектура позволяет нам создавать несколько независимых частей приложения для работы с одним и тем же событием и выполнения различных целенаправленных задач. Это может быть выгодно по нескольким причинам:

• Команда должна сосредоточиться только на одной части.
• Код приложения для мелких деталей может быть простым.

Еще одно преимущество этой концепции заключается в том, что она позволяет нам предоставить пользователю быстрый “интерфейс”. В качестве примера вышеупомянутого приложения, пользователю необходимо разместить заказ. Приложение может продолжать обрабатывать все остальные неинтерактивные задачи, такие как обновление инвентаря и взаимодействие с приложением доставки, без внимания пользователя.

Это также позволяет легко добавлять новые этапы обработки в конвейер событий. Например, допустим, нам нужно, чтобы из события выполнялась дополнительная задача. Нам просто нужно добавить нового потребителя для обработки события, не затрагивая никакую другую часть приложения.

Наконец, гораздо проще масштабировать каждый отдельный модуль, если он отделен от других, чем масштабировать все приложение вместе. Это еще более выгодно, когда есть часть, которая потребляет гораздо больше ресурсов, чем ее аналоги в конвейере обработки событий.

Однако событийно-управляемая архитектура не лишена недостатков, если она недостаточно хорошо продумана. Если применять его к очень простым проблемам, это может привести к сложным рабочим процессам, которые медленно и трудно отлаживать.

Давайте рассмотрим несколько простых способов реализации событийно-управляемой архитектуры с помощью Elixir без необходимости написания сложных фрагментов кода.

Синхронная событийно-управляемая архитектура в Elixir

Самым простым (и неэффективным) способом выполнения вышеописанного потока было бы выполнение всего синхронно по запросу пользователя.

Так, если у вас есть модуль Orders, который обрабатывает запрос пользователя, синхронная реализация может выглядеть следующим образом:

defmodule Orders do
  def create_order(attrs) do
    # salva o pedido
    {:ok, order} = save_order(attrs)
    # atualiza o estoque
    {:ok, _inventory} = update_inventory(order)
    # cria a entrega
    {:ok, _delivery} = create_delivery(order)
    # retorna o pedido
    {:ok, order}
  end
end

Мы можем улучшить это, чтобы потребители новых событий могли легче масштабироваться:

defmodule Orders do
  @event_consumers [
    {Inventory, :handle_event},
    {Delivery, :handle_event},
  ]

  def create_order(attrs) do
    {:ok, order} = save_order(attrs)

    event = %Orders.Event{type: :new_order, payload: order}
    @event_consumers
    |> Enum.each(fn {module, func} ->
      apply(module, func, [event])
    end)

    {:ok, order}
  end
end

При такой реализации для добавления новых потребителей достаточно добавить спецификацию в список @event_consumers, и эти потребители могут работать независимо.

Хотя синхронный подход хорошо работает для небольшого числа потребителей, у него есть недостаток. Создание заказа может занять много времени, так как вам придется ждать обновления запасов и создания доставки.

Это внутренние задачи, которые могут быть выполнены без взаимодействия с пользователем и перемещены из синхронной цепочки.

Давайте посмотрим, как можно еще больше оптимизировать событийно-управляемую архитектуру с помощью GenServer.

Использование GenServer в событийно-управляемой архитектуре в Elixir

В данной реализации мы запустим отдельный процесс для каждого потребителя. Они подписываются на событие производителя и выполняют свои задачи одновременно.

Для шины событий мы можем использовать Phoenix.PubSub. Обратите внимание, что приложения, не использующие Phoenix, могут напрямую использовать реестр в качестве PubSub.

Во-первых, давайте посмотрим на производителя.

defmodule Orders do
  def create_order(attrs) do
    {:ok, order} = save_order(attrs)
    event = %Orders.Event{type: :new_order, payload: order}
    Phoenix.PubSub.broadcast(:my_app, "new_order", event)
    {:ok, order}
  end
end

В новом заказе (порядке) мы создаем структуру события и используем Phoenix.PubSub.broadcast/3 для трансляции этого события на шину. Как видите, это намного проще, чем предыдущая реализация, где модуль Orders обрабатывал задания от другого модуля последовательно.

Затем потребители могут подписаться (subscribe) на поток new_order и реализовать handle_info/2, чтобы получать уведомления каждый раз, когда производитель публикует новое событие.

defmodule Inventory do
  use GenServer

  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)

  def init(_opts) do
    Phoenix.PubSub.subscribe(:my_app, "new_order")
  end

  def handle_info(%Orders.Event{type: :new_order, payload: order}, state) do
    state = consume(state, order.product)
    {:noreply, state}
  end
end

Модуль Delivery будет очень похож на описанный выше, поэтому здесь я его пропускаю.

Как вы можете видеть, это намного лучше, чем предыдущие реализации. Модули Inventory и Delivery могут применяться независимо к потоку new_order. Модуль Orders передает в эту тему сообщения о новых заказах, и события доставляются в подписанные процессы.

Вы даже можете распределить это между несколькими узлами и Phoenix.PubSub (с помощью PG, Redis или другого адаптера), распространяя события на все узлы.

Отлично, правда? Вообще-то, нет. Существует несколько проблем с этим подходом:

• PubSub обеспечивает вещание в реальном времени без очереди сообщений, поэтому если один из процессов подписчика находится в режиме ожидания, он может пропустить вещание.

• Если подписчик выполняет какую-то тяжелую работу, он может не успевать за входящими сообщениями, что приведет к тайм-ауту и, как следствие, к краху дерева процессов.

• Если при обработке сообщения абонент сталкивается с ошибкой, оно считается использованным и не будет обработано (повторно) позже.

Поэтому для нашего текущего случая использования этот подход не подходит.

Тем не менее, у этого подхода все еще есть свои варианты использования. Его можно использовать для задач, которые не являются критическими или могут быть исправлены в следующем сообщении: например, если задача вычисляет предложения о следующих покупках пользователя на основе его последней покупки. Хотя это также должно срабатывать при новом заказе, это не совсем критично (для традиционного сайта электронной коммерции) и может пересчитать предложения для пользователя при его следующей покупке.

Событийно-управляемая реализация с использованием GenStage в Elixir

В предыдущем разделе мы увидели отличную реализацию нашей событийно-управляемой системы с использованием GenServer. Но это не обошлось без ограничений. Посмотрим, как проявит себя GenStage.

GenStage делает четкое различие между производителями, потребителями и производитель_потребитель, и каждый процесс должен выбрать один из них при запуске (в init/1). В нашем случае и Inventory, и Delivery являются потребителями, а Orders – производителем.

Здесь начинаются некоторые сложности. В GenStage есть понятие спроса. Каждый потребитель может выдать запрос на то, сколько событий он может обработать. производитель должен отправить эти события потребителю. Давайте посмотрим на базовый producer в действии.

defmodule Orders do
  use GenStage

  def start_link(opts) do
    GenStage.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def init(_opts) do
    {:producer, :some_state_which_does_not_currently_mattere}
  end

  def create_order(pid, attrs) do
    GenStage.cast(pid, {:create_order, attrs})
  end

  def handle_cast({:create_order, attrs}, state) do
    {:ok, order} = save_order(attrs)
    {:noreply, [%Orders.Event{type: new_order, payload: order}], state}
  end

  def handle_demand(_demand, state), do: {:noreply, [], state}
end

Основная часть нашего кода находится в handle_cast, где мы храним порядок (order) и возвращаем кортеж в виде {:noreply, events, new_state}. Новые события хранятся во внутреннем буфере GenStage и отправляются потребителям по мере того, как они делают новые запросы (или немедленно, если есть потребители с неудовлетворенным спросом).

Давайте посмотрим пример реализации consumer:

defmodule Inventory do
  use GenStage

  def start_link(opts) do
    GenStage.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def init(_opts) do
    {:consumer, [], subscribe_to: [Orders]}
  end

  def handle_events(events, _from, state) do
    state = Enum.reduce(events, state, & handle_event(&1, &2))
    {:noreply, [], state}
  end

  def handle_event(%Orders.Event{type: :new_order, payload: order}, state) do
    new_state = update_inventory(order)
    new_state
  end
end

В consumer он сначала предупреждает, что у нас есть subscribe_to (subscribe_to) внутри init/1. Это автоматически подписывает/подписывается (subscribe) Inventory на любое событие, опубликованное Orders. Дополнительные опции, доступные в init, см. в документации GenStage.

Здесь большая часть работы происходит внутри handle_events/3, который автоматически вызывается GenStage, как только становятся доступны новые события. Здесь мы имеем дело с событием new_order, обновляющим inventory и возвращающим новое состояние.

При такой простой реализации мы получаем несколько преимуществ, превосходящих реализацию GenServer:

• Автоматическое сохранение событий во внутреннем буфере GenStage, когда producer имеет новые события без доступного consumer.

Даже если потребители отсутствуют во время производства некоторых событий, мы все равно гарантированно получим их, когда потребитель снова станет доступен.

Ознакомьтесь с руководством Genstage по буферизации спроса для усовершенствованной логики буферизации.

• Автоматическое распределение работы между несколькими потребителями.

Если у вас есть задачи с большим количеством потребителей, вы можете запустить несколько потребительских процессов. DemandDispatcher по умолчанию для GenStage будет равномерно распределять работу между всеми процессами.

Смотрите GenStage.Dispatcher для других стратегий диспетчеризации для распределения событий среди всех потребителей или распределения разделов среди потребителей на основе хэш-функции.

Но, как и в случае с GenServer или синхронной реализацией, использование GenStage не обходится без проблем.

Если consumer падает во время обработки события, GenStage будет считать событие доставленным и не отправлять его снова, когда consumer снова станет доступен.

Чтобы убедиться, что вы правильно отслеживаете сбои, вы можете использовать службу мониторинга, например AppSignal. AppSignal [легко устанавливается для вашего Elixir-приложения] и помогает вам контролировать производительность, а также отслеживать ошибки. Вот пример приборной панели ошибок, которую предоставляет AppSignal:

Вы также можете настроить уведомления о сбоях через AppSignal.

На стороне приложения вы можете хранить такие события в постоянном магазине после того, как они будут доставлены потребителю. Если потребитель терпит крах, то после восстановления он может вернуться к кэшированным событиям.

Однако будьте очень осторожны и не производите слишком много мероприятий без достаточного количества потребителей. Хотя GenStage предлагает автоматический буфер событий, этот буфер имеет максимальный размер (настраиваемый) и практический максимальный размер, ограниченный памятью сервера.

Если вы не контролируете частоту создания событий, рассмотрите возможность использования внешнего хранилища данных, такого как Redis или Postgres, для буферизации событий.

Заключение: Event-Driven Architecture в Elixir – выход за пределы GenStage

В этом посте мы рассмотрели три подхода к реализации событийно-управляемой системы в Elixir: синхронно, с использованием GenServer и, наконец, с использованием GenStage. Мы рассмотрим некоторые преимущества и недостатки каждого подхода.

Простой пример GenStage может стать отправной точкой для реализации сложных конвейеров обработки данных, управляемых событиями, охватывающих несколько узлов. Для получения дополнительной информации я предлагаю вам ознакомиться с замечательной документацией GenStage.

Если вы ищете еще большую абстракцию, Бродвей – хорошее место для начала. Он построен поверх GenStage и предлагает несколько дополнительных функций, включая потребление данных из внешних очередей, таких как Amazon SQS, Apache Kafka и RabbitMQ.

До следующего раза, счастливого кодирования!