«Будем отправлять события в 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 для решения задач асинхронной обработки и коммуникаций с другими сервисами. Целью же этой статьи было дать общий обзор темы событий и сообщений. Надеюсь, было полезно и интересно, спасибо что дочитали.)