«Будем отправлять события в Rabbit!» Фраза, которая выдает мышление, рождающее код, полный боли. К сожалению, я ее часто слышу. Поэтому, уже много лет размышлял о написании этой статьи и безумно рад, что у меня, наконец, дошли до нее руки.
В статье я расскажу, как смешение понятий события, сообщения и транспорта рождает возгласы типа «Я ненавижу использовать Symfony Messenger, потому что был у меня проект на нем, и он не взлетел!»
Будут косвенно затронуты компоненты Symfony Messenger и Event Dispatcher. Несмотря на это, данный материал может оказаться полезным и для разработчиков, использующих другие фреймворки и даже другие языки.
Что такое события
Давайте начнем с событий. Если говорить в узком контексте разработки приложения, то событие (Event)- это определенная структура данных (чаще всего объект), которая несет в себе информацию о деталях произошедшего. Например:
readonly class UserCreatedEvent extends Event
{
public function __construct(
public string $userId
) {
}
}На самом деле, объект события может вообще не иметь полей. Сам факт появления объекта класса UserCreatedEvent - уже информация о том, что был создан новый пользователь, и этой информации может быть достаточно, например, для отправки простейшего уведомления админу.
К слову, если вам интересно, я недавно сравнивал плюсы и минусы трех способов выражения разновидностей объектов: через наследование (как в примере выше), через поле type (для событий тоже может быть актуальным) и через композицию (для событий не используется).
Каково назначение событий? Напомню, мы сейчас рассуждаем в узком контексте разработки приложения. Так вот, в этом контексте событие - это способ ветвления кода. Мы могли бы в месте выброса события сразу поместить весь код, который должен быть вызван в качестве реакции на это событие: отправить уведомления, дернуть какие-то куски бизнес-логики и так далее.
Но, сами понимаете, при таком подходе наш код превратится в известную субстанцию. Поэтому и были придуманы события, а точнее, паттерн Observer. Так вот, чтобы не писать весь код реакции на событие в виде простыни, помещаемой в место возникновения события, придумали обработчики события, они же handlers, listeners, subscribers, кому как угодно.
В проекте, построенном с применением слоистой архитектуры и DDD (или хотя бы чего-то одного из двух), важно понимать, какие виды событий вы будете использовать. Точнее, в каких слоях будут генерироваться ваши события и в каких - обрабатываться.
Давайте остановимся на этом чуть поподробнее.
Symfony Event Dispatcher и доменный Event Dispatcher
В Симфони есть такой компонент, как Event Dispatcher, который предоставляет весь необходимый функционал для публикации событий и их обработки. Причем, как и остальные компоненты Symfony, Event Dispatcher можно использовать и в проектах, построенных на других фреймворках.
Несмотря на то, что этот компонент дает нам весь необходимый функционал для работы с событиями, возникает вопрос, везде ли мы можем его использовать? Правильный ответ - нет. Symfony Event Dispatcher - это инфраструктурный компонент. А значит и использоваться он может только в инфраструктурном слое.
Для работы с доменными событиями будет правильным иметь собственный - доменный - Event Dispatcher. Набросать его - дело пяти минут:
namespace App\Domain\Event;
class DomainEventPublisher
{
/** @var DomainEventSubscriberInterface[] */
private array $subscribers = [];
private static ?static $instance = null;
public static function instance(): static
{
if (null === static::$instance) {
static::$instance = new static();
}
return static::$instance;
}
public function __clone()
{
throw new \BadMethodCallException('Clone is not supported');
}
public function subscribe(DomainEventSubscriberInterface $subscriber): void
{
$this->subscribers[] = $subscriber;
}
public function publish(DomainEvent $event): void
{
foreach ($this->subscribers as $subscriber) {
if ($subscriber->isSubscribedTo($event)) {
$subscriber->handle($event);
}
}
}
}К слову, идея такого DomainEventPublisher позаимствована мной из книги Domain-Driven Design in PHP. Как видите, этот класс довольно прост: реализуя Singleton, он сохраняет в себя все объекты-подписчики, реализующие DomainEventSubscriberInterface, а при публикации доменного события просто перебирает всех своих подписчиков и дергает тех из них, кто подписан на публикуемое событие.
Код DomainEventSubscriberInterface и DomainEvent и того проще:
abstract class DomainEvent
{
}
interface DomainEventSubscriberInterface
{
public function isSubscribedTo(DomainEvent $event): bool;
public function handle(DomainEvent $event): void;
}Таким образом, мы получили в своем доменном слое простейшую реализацию паттерна Observer и удобный способ ветвления кода. Теперь мы можем в любом месте доменного слоя выбрасывать события и реализовывать подписчиков на них. Причем, подписчики могут размещаться уже как в доменном, так и в инфраструкурном слоях. Давайте рассмотрим простейший пример.
namespace App\Domain\Model\User;
use App\Domain\Event\DomainEventPublisher;
use App\Domain\Event\User\UserCreated;
class User
{
public function __construct(string $id, string $name)
{
//...
DomainEventPublisher::instance()->publish(new UserCreated($id));
}
}Теперь мы можем писать различные обработчики для события UserCreated. Например, обработчик отправляющий в telegram приветствие нового пользователя, мы поместим в инфраструктурный слой, а в доменном слое мы могли бы разместить обработчики, дергающие бизнес-логику из других bounded contexts.
К слову, я настоятельно рекомендую до последнего избегать размещения любой логики в обработчиках событий, потому что обработчики событий крайне неудобно дебажить. Чтобы выполнение обработчика не прошло мимо вас во время отладки, вы должны знать, что этот обработчик будет вызван и заранее поставить в нем точку останова. В противном случае вы столкнетесь с классическим проявлением магии: в поведении и данных вашего приложения произойдут изменения, источник которых останется для вас непонятен.
Еще один важный момент. При работе с доменными событиями есть правило: выбрасывайте событие как можно ближе к месту возникновения. Если наш класс User - это доменный агрегат (а оно так обычно и бывает), то он может агрегировать в себя класс Profile, а тот, в свою очередь агрегировать в себя класс Contacts. Так вот, событие UserContactChanged нужно выбрасывать не в классе User и не в классе Profile, а непосредственно в классе Contacts в тех методах, где происходит изменение контактов.
Такой подход исключит ситуации, когда событие де-факто произошло, а дернуть DomainEventPublisher забыли. К слову, если вам интересно узнать немного больше о доменных агрегатах, то можете ознакомиться с моими постами про базовое правило построения доменного агрегата и о том, что такое инвариант
Ну а пока давайте подведем промежуточный итог.
Событие в контексте разработки приложения - это структура данных, несущая информацию о деталях произошедшего.
Событие - это способ ветвления кода. Мы используем обработчики событий, чтобы аккуратно разложить по классам и слоям код реакции на эти события.
В доменном слое мы не можем использовать инфраструктурные компоненты, такие как Symfony Event Dispatcher. Поэтому нам понадобится собственный простейший
DomainEventPublisherИспользуйте Symfony Event Dispatcher для работы со встроенными событиями фреймворка в инфраструктурном слое.
Обработчики событий крайне неудобно дебажить. Поэтому, старайтесь избегать использовать события для реализации бизнес-логики. Также старайтесь избегать создания большого количества кастомных инфраструктурных событий и их обработчиков.
Используйте события для решения попутных задач, таких как отправка уведомлений, логирование, scheduling тяжеловесных задач с отложенным их выполнением и т. д.
Размещайте код публикации события как можно ближе к месту возникновения события.
Теперь, когда мы разобрались с событиями и Symfony Event Dispatcher, давайте перейдем к сообщениям и Symfony Messenger.
Отличия сообщений от событий
Сообщение, если придерживаться все того же узкого контекста разработки приложений - это способ коммуникации между приложениями. Коммуникация между приложениями бывает синхронной и асинхронной, а так же один к одному и один ко многим. Подробнее об этом можно прочитать в книге Криса Ричардсона "Микросервисы".
Так вот, когда мы взаимодействуем с другим приложением или сразу несколькими приложениями, мы отправляем сообщение. Не событие, а именно сообщение (которое в том числе может быть и о событии). При этом, когда мы говорим об отправке сообщений другим приложениям (сервисам и микросервисам), мы верхнеуровнево рассуждаем о том, будет эта отправка синхронной или асинхронной, и будет у этого сообщения один получатель, или много. И мы ничего на этапе проектирования не говорим о транспорте (Rabit, Kafka и т. д.)
Давайте в этой терминологии рассмотрим вызов API эндпоинта стороннего сервиса с вот таким телом запроса:
{
"event": {
"type": "passport.user.created",
"timestamp": 1744875021,
"data": {
"userId": 123,
"userName": "Vasya"
}
}
}Об этом вызове мы можем сказать, что мы отправили синхронное сообщение один к одному о событии passport.user.created, используя транспорт HTTP. Мы можем взять тот же самый JSON и отправить его в RabbitMQ на topic exchange, задав routing key passport.user.created. В этом случае мы скажем, что мы отправили асинхронное сообщение один ко многим о событии passport.user.created, используя транспорт RabbitMQ.
Теперь давайте вернемся к написанному в предыдущих разделах и попробуем собрать все воедино.
Во время выполнения нашего приложения могут возникать события.
Мы создаем в коде события для того, чтобы не писать весь код реакции на событие в месте его возникновения.
События могут возникать как в доменном, так и в инфраструктурном слое нашего приложения.
Обработчики событий, содержащие код реакции на события, также могут принадлежать как к доменному, так и к инфраструктурному слою приложения, в зависимости от того, какую задачу они решают.
Если нам необходимо отправить в другие сервисы информацию о возникшем в нашем приложении событии, нам понадобится две вещи: сообщение и транспорт.
Внутрь сообщения мы помещаем информацию о событии. Ничто не мешает нам положить в одно сообщение информацию о нескольких событиях, или дополнительную информацию, не связанную с событием.
При проектировании обмена сообщениями между сервисами важно правильно выбрать способ взаимодействия: один к одному или один ко многим и синхронный или асинхронный.
В самую последнюю очередь, после того, когда вы уже знаете, о чем будут ваши сообщения и каким способом они будут отправляться, вы должны выбрать наиболее подходящий транспорт. Именно способ отправки сообщения определяет транспорт, а не наоборот, как, к сожалению, происходит на большинстве проектов. Ведь с помощью RabbitMQ можно отправлять и синхронные один к одному сообщения: достаточно заблокировать отправляющий сервис до момента получения ответного сообщения. Другой вопрос: захотите ли вы использовать RabbitMQ таким способом.
Основная мысль здесь состоит в том, что вы сначала определяетсь с тем, что вы отправляете (событие, документ, команду), кому (одному или многим) вы отправляете и как (синхронно или асинхронно) вы это делаете. А уже потом вы подбираете транспорт, а после выбора транспорта вы подбираете инфраструктурный компонент вашего приложения, который позволит вам взаимодействовать с этим транспортом (например, Symfony Messenger).
Этот подход намного лучше такого, когда объявляется: "Будем использовать Symfony Messenger, чтобы отправлять события в Rabbit!" Применение такого подхода отражается и в коде: если у вас один класс - это и событие и сообщение, а второй класс - это одновременно ответ на сообщение и обработчик события, то проблем вам не избежать. Отсюда и рождается мнение, что "Symfony Messenger плохой".
Symfony Messenger - всего лишь инструмент для взаимодействия с транспортом. И он должен знать свое место и решать только те задачи, для которых создан, а не быть вместе с кроликом "центром вселенной".
Вот мы и подошли к заключительному разделу о месте в нашем приложении событий, их обработчиков и транспортов для отправки и получения сообщений о событиях.
Преобразование нашего события в исходящее сообщение и обработка ответа
Итак, давайте теперь схематично разложим, как события и сообщения соотносятся друг с другом в слоистой архитектуре. Для начала давайте взглянем на вот такую упрощенную схемку
В доменном слое у нас сущность User, наш DomainEventPublisher, который мы используем для публикации событий и событие UserCreatedEvent. У этого события есть два обработчика.
Первый - RefferalUserCreatedSubscriber - расположен в доменном слое и тригерит бизнес-логику связанного контекста (bounded context). Например, этот обработчик может вызывать бизнес-логику биллинга для начисления вознаграждения юзеру, который пригласил вновь созданного пользователя.
Второй обработчик - BusMessageConverter - расположен в инфраструктурном слое. Его задача перелить данные из доменного UserCreatedEvent в инфраструктурный BusMessage и отправить его в транспорт с помощью Symfony Messenger.
Теперь давайте посмотрим на схему, которая изображает процесс обработки входящего сообщения
Symfony Messenger, получая сообщение из транспорта, декодирует его с помощью собственного сериалайзера (не того, который Symfony Serializer). Результатом декодирования становится инстанс BusMessage в инфраструктурном слое.
BusMessageHandler является обработчиком BusMessage. Этот класс - именно обработчик в системе Symfony Messenger (помечен соответствующим тегом или атрибутом). BusMessageHandler выступает, по сути, контроллером, который обрабатывает входящее сообщение. Также, как привычные нам контроллеры обрабатывают HTTP запросы. Разница здесь только в виде транспорта и возможной (но не обязательной) асинхронности, предоставляемой SymfonyMessenger и транспортом, который он использует.
Ну а поскольку BusMessageHandler является обычным контроллером, то и работать с ним можно и нужно как с обычным контроллером. Например, инстанцировать ApplicationCommand, которая уже дернет необходимую бизес-логику. Например, изменит статус юзера и сохранит изменения в репозитории.
Последняя часть схем, где Symfony Messenger, изображена очень поверхностно и условно. Symfony Messenger состоит из нескольких компонентов, и его грамотное использование сопряжено с решением сразу нескольких задач.
В планах статья, где я расскажу более детально именно об использовании Symfony Messenger для решения задач асинхронной обработки и коммуникаций с другими сервисами. Целью же этой статьи было дать общий обзор темы событий и сообщений. Надеюсь, было полезно и интересно, спасибо что дочитали.)
