Проектируем реактивное — Message-Driven системы на PHP

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

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

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

В достижении этого нам может помочь реактивный манифест.

Цель этой статьи — показать не только зачем, но и как мы можем создавать приложения на PHP, которые будут устойчивыми, масштабируемыми и податливыми для  изменений. Статья концентрируется на разработке  бизнес-ориентированных приложений, в которых можно четко выделить бизнес-логику, процессы и потоки работ (workflows). Это квинтэссенция моего опыта, накопленного за годы работы над бизнес-ориентированным программным обеспечением в целом и в процессе создания фреймворка для обмена сообщениями Ecotone в частности.

Реактивные системы

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

Реактивными можно назвать системы, которые являются отзывчивыми (Responsive), отказоустойчивыми (Resilient), эластичными (Elastic) и ориентированными на передачу сообщений (Message Driven).

Реактивный манифест раскрывает каждый из этих четырех принципов.

Слово «реактивные», используемое здесь, может натолкнуть вас на мысли о реактивном программировании, но это не совсем одно и то же. Реактивные системы требуют соблюдения определенных принципов проектирования при создании и интеграции приложений, а реактивное программирование является парадигмой. Если вам интересно узнать об этом больше, вы можете прочитать эту статью Джонаса Бонера, автора Реактивного манифеста.

Давайте озвучим каждый из этих принципов.

• Отзывчивость (Responsiveness) подразумевает способность своевременно реагировать и предоставлять пользователям положительный опыт, что в результате побуждает их продолжать использовать приложение. Этого качества можно добиться множеством способов, среди которых можно выделить: специальные наблюдаемые модели, кэширование, масштабирование и т. д.

• Эластичность (Elasticity) подразумевает возможность масштабировать систему в соответствии с текущими потребностями.
  Когда нагрузка возрастает, а скорость отклика падает, мы подключаем больше серверов или потребителей. С другой стороны, когда падает нагрузка, мы бы хотели уменьшить масштаб, чтобы высвободить ресурсы и снизить затраты.

• Отказоустойчивость (Resiliency) требует быть готовыми к сбоям. Когда нам приходится полагаться на какие-нибудь внешние по отношению к нашему приложению компоненты (а это может быть даже другое приложение в локальной сети), они могут давать сбой. Быть отказоустойчивым — означает понимать, что сбои будут происходить, но мы можем предвидеть их и внедрить в архитектуру приложения решения, которые помогут ему восстановиться.

• Ориентированность на коммуникацию посредством сообщений (Message Driven) означает, что компоненты, модули и сервисы взаимодействуют между собой посредством передачи сообщений, а не прямых вызовов. Сообщения — это фрагменты данных, которые можно передавать по сети. Они идентифицируемы и несут в себе намерение. Когда сообщение отправлено, другие компоненты могут потребить его в какой-то момент в будущем. Тем не менее, между компонентами нет прямой зависимости, и они не полагаются на доступность друг друга. Устраняю прямую зависимость между компонентами, мы делаем их естественным образом изолированными, благодаря чему они остаются незатронутыми сбоями друг-друга.

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

Например, Message-Driven системы позволяют масштабировать потребителей и регулировать поток сообщений. Такие системы по своей природе должны быть масштабируемыми и отзывчивыми.

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

Создание реактивных систем на PHP

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

Разобравшись с теорией, пришло время закатать рукава и провести рефакторинг некоторого кода на PHP, который не задействует Message-Driven архитектуру.

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

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

Для доставки продуктов клиентам мы интегрируемся с внешней службой доставки.

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

• Внесение заказа в базу данных.

• Отправка клиенту электронного письма с подтверждением и кратким описанием заказа.

• Запуск процесса доставки путем вызова службы доставки (Shipping Service) через HTTP API

Этап 1 — Понимание проблемы

Давайте рассмотрим следующую реализацию приведенного выше сценария:

<?php

class OrderController
{
    public function __construct(private AuthenticationService $authenticationService, private OrderService $orderService) {}

    public function placeOrder(Request $request): Response
    {
        $data = \json_decode($request->getContent(), true, flags: JSON_THROW_ON_ERROR);
        $currentUserId = $this->authenticationService->getCurrentUserId();
        $shippingAddress = new ShippingAddress($data['address']['street'], $data['address']['houseNumber'], $data['address']['postCode'], $data['address']['country']);
        $productId = Uuid::fromString($data['productId']);

        $this->orderService->placeOrder($currentUserId, $shippingAddress, $productId);

        return new Response();
    }
}

class OrderService
{
    public function __construct(
        private OrderRepository   $orderRepository, private UserRepository $userRepository,
        private ProductRepository $productRepository, private NotificationSender $notifcationSender,
        private ShippingService   $shippingService, private Clock $clock
    ) {}

    /**
     * @param UuidInterface[] $productIds
     */
    public function placeOrder(UuidInterface $userId, ShippingAddress $shippingAddress, array $productIds): void
    {
        $productDetails = $this->productRepository->getBy($productId)->getProductDetails();
    
        /** Внесение заказа в базу данных /
        $order = Order::create($userId, $shippingAddress, $productDetails, $this->clock);
        $this->orderRepository->save($order);

        /** Отправка уведомления о подтверждении заказа */
        $user = $this->userRepository->getBy($order->getUserId());
        $this->notifcationSender->send(new OrderConfirmationNotification(
            $user->getFullName(), $order->getOrderId(), $productDetails, $order->getTotalPrice())
        );

         /** Вызов службы доставки через HTTP для доставки товаров */
        $this->shippingService->shipOrderFor($userId, $order->getOrderId(), $productDetails, $shippingAddress);
    }
}

Пример потока вызова службы доставки с побочными действиями 

В рамках оформления заказа происходит пара вещей.

Мы имеем дело с данными (отражение заказа в базе данных) и с побочными действиями (отправка электронной почты и вызов службы доставки).

Если один из компонентов выходит из строя (например, отправка электронного письма), это может повлиять на другие части (доставка или сохранение заказа).

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

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

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

Существует другой архитектурный подход, при котором подобные проблемы решаются на уровне проектирования: Message-Driven системы.

Этап 2 — Пересаживаем наш код на сообщения, делаем его отказоустойчивым

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

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

Message-Driven архитектура нуждается в абстракции, специально построенной поверх языка. Фреймворк NServiceBus обеспечивает реализацию архитектуры обмена сообщениями на языке C#, проект Spring Integration для Spring Cloud реализует обмен сообщениями на Java, ну а для PHP существует фреймворк Ecotone.

Термин “Ecotone” (экотон) в экологии означает переходную зону между экосистемами, такими как, например, лес и луг.

Фреймворк Ecotone функционирует как переходная зона между вашими компонентами, модулями и сервисами. Он соединяет все между собой, но уважает границы каждой области.

Ecotone’s Message contains a Payload and a Headers:

Сообщение Ecotone содержит полезную payload и заголовки:

<?php

interface Message
{
    public function getHeaders(): MessageHeaders;

    public function getPayload(): mixed;
}

Интерфейс Message Ecotone

Payload-ом может быть что угодно, будь то строка json/xml, инстанс объекта, массив и т. д.

Заголовки — это метаданные сообщения, содержащие информацию, нужную фреймворку, и пользовательскую информацию, которая не является частью payload-а (например, таймстемп, исполнителя сообщения, роли исполнителя).

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

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

Давайте двинемся дальше и определим наши Command Message и Event Message, которые помогут нам вычленить побочные побочные действия.

<?php
/** Команда */
final class PlaceOrder
{
    /** @param UuidInterface[] $productIds */
    public function __construct(
      public readonly UuidInterface $userId,
      public readonly ShippingAddress $address,
      public readonly UuidInterface $productId
    ) {}
}

/** Событие */
final class OrderWasPlaced
{
    public function __construct(
        public readonly UuidInterface $orderId
    ) {}
}

Классы Command и Event

Мы изменим наш класс OrderService, чтобы он мог получать команды (Command) и публиковать события (Event).

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

На самом деле это концепции более высокого уровня, которые делают намерение сообщения явным. В Ecotone они представлены POPO (Plain Old PHP Objects).

Ecotone заботится о сериализации и десериализации payload-ов и заголовков, когда они генерируются.

<?php

final class OrderService
{
    public function __construct(
        private OrderRepository $orderRepository, private ProductRepository $productRepository,
        private Clock           $clock, private EventBus $eventBus
    ) {}

    #[CommandHandler]
    public function placeOrder(PlaceOrder $command): void
    {
        $productDetails = $this->productRepository->getBy($command->productId)->getProductDetails();

        /** Сохранение заказа в базе данных */
        $order = Order::create($command->userId, $command->shippingAddress, $productDetails, $this->clock);
        $this->orderRepository->save($order);

        /** Публикация события, указывающего, что заказ был оформлен */
        $this->eventBus->publish(new OrderWasPlaced($order->getOrderId()));
    }
}

Обработка команды и публикация события о том, что заказ был оформлен.

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

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

С помощью атрибута #[CommandHandler] мы указываем, что этот метод отвечает за обработку команды PlaceOrder. Ecotone делает вывод об этом на основании типа первого типизированного свойства.

И вот как мы вызываем этот обработчик команд (Command Handler) из контроллера, который теперь использует шину команд (Command Bus):

<?php

final class OrderController
{
    public function __construct(private AuthenticationService $authenticationService, private CommandBus $commandBus) {}

    public function placeOrder(Request $request): Response
    {
        $data = \json_decode($request->getContent(), true, flags: JSON_THROW_ON_ERROR);
        $currentUserId = $this->authenticationService->getCurrentUserId();
        $shippingAddress = new ShippingAddress($data['address']['street'], $data['address']['houseNumber'], $data['address']['postCode'], $data['address']['country']);
        $productId = Uuid::fromString($data['productId']);

        $this->commandBus->send(new PlaceOrder($currentUserId, $shippingAddress, $productId));

        return new Response();
    }
}

Вызов шины команд из контроллера

Если вы используете Symfony или Laravel, Ecotone автоматически зарегистрирует интерфейсы шины команд и событий в вашем контейнере зависимостей.

Давайте определим обработчики событий, которые подписываются на событие OrderWasPlaced и запускают наши побочные действия:

<?php

final class NotificationSubscriber
{
    #[Asynchronous("asynchronous_channel")]
    #[EventHandler(endpointId: "notifyWhenOrderWasPlaced")]
    public function whenOrderWasPlaced(OrderWasPlaced $event, OrderRepository $orderRepository, UserRepository $userRepository, NotificationSender $notificationSender): void
    {
        /** Отправка уведомления о подтверждении заказа */
        $order = $orderRepository->getBy($event->orderId);
        $user = $userRepository->getBy($order->getUserId());

        $notificationSender->send(new OrderConfirmationNotification($user->getFullName(), $order->getOrderId(), $order->getProductDetails(), $order->getTotalPrice()));
    }
}

final class ShippingSubscriber
{
    #[Asynchronous("asynchronous_channel")]
    #[EventHandler(endpointId: "shipWhenOrderWasPlaced")]
    public function whenOrderWasPlaced(OrderWasPlaced $event, OrderRepository $orderRepository, ShippingService $shippingService): void
    {
        /** Отправка уведомления о подтверждении заказа */
        $order = $orderRepository->getBy($event->orderId);

        /** Вызов службы доставки через HTTP для доставки товаров */
        $shippingService->shipOrderFor($order->getUserId(), $order->getOrderId(), $order->getProductDetails(), $order->getShippingAddress());
    }
}

Обработка события OrderWasPlaced двумя обработчиками событий

Мы определили обработчики событий, которые подписываются на событие OrderWasPlaced.

Ecotone использует концепцию, называемую «каналы сообщений» (Message Channels).

Каналы сообщений подобны конвейерам, по которым проходят сообщения. В приведенном выше примере канал называется “asynchronous_channel”.

Если мы используем шину команд и событий, то нам понадобится и шлюз обмена сообщениями (Messaging Gateway).

Он берет наши данные и преобразует их в сообщение Ecotone. Именно через этот простой интерфейс мы и включаемся в обмен сообщений.

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

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

Поскольку мы определили обработчики событий как асинхронные, с помощью атрибута 

#[Asynchronous(“asynchronous_channel”)], события будут обрабатываться асинхронно.

В зависимости от ваших потребностей существует несколько реализаций канала сообщений: RabbitMQ, SQS, Redis и т.д. Тем не менее, какое бы решение ни было выбрано, на бизнес-код это не повлияет.

Больше информации об обработке команд вы можете найти в документации Ecotone.

Если вы хотите узнать больше об обработке событий, почитайте этот раздел документации Ecotone.

Защита данных и побочных действий от сбоев

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

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

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

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

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

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

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

Подробнее об асинхронных обработчиках событий вы можете почитать в этом посте в блоге Ecotone.

Подробнее о реализации паттерна издатель-подписчик с обработчиками событий вы можете почитать в документации Ecotone.

Защита ваших сообщений от потери

Мы еще не до конца обезопасили формирование заказа.

После сохранения заказа мы публикуем события во внешнем брокере сообщений. Таким образом задействовано более одного механизма хранения (база данных и брокер сообщений).

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

Хорошим решением в таких случаях является паттерн Outbox (“папка с исходящими”), и Ecotone реализует его с помощью каналов сообщений.

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

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

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

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

<?php

final class MessageChannelConfiguration
{
    #[ServiceContext]
    public function asynchronousChannel()
    {
        return DbalBackedMessageChannelBuilder::create("asynchronous_channel");
    }
}

Ecotone позволяет добавить дополнительную конфигурацию с помощью ServiceContext.

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

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

Подробнее об использовании паттерна Outbox и масштабировании решения можно почитать в этом посте в блоге Ecotone.

Обработка дедупликации сообщений

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

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

Все сообщения Ecotone содержат уникальный MessageId, по которому их можно идентифицировать и отслеживать. На основе этого, Ecotone уже предоставляет механизм дедупликации, реализующий идемпотентного потребителя (idempotent consumer).

Когда сообщение успешно обработано, Ecotone сохраняет MessageId. Таким образом можно отслеживать повторяющиеся сообщения и, при необходимости, отбрасывать их.

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

<?php

final class WebhookController
{
    public function __construct(private CommandBus $commandBus) {}

    public function placeOrder(Request $request): Response
    {
        $data = \json_decode($request->getContent(), true, flags: JSON_THROW_ON_ERROR);

        $this->commandBus->sendWithRouting("webhook.store", $data, metadata: ["eventId" => $data['eventId']]);

        return new Response();
    }
}

final class WebhookHandler
{
    #[Deduplicated("eventId")]
    #[CommandHandler("webhook.store")]
    public function handle(array $payload): void
    {
        // сохраняем его
    }
}

Дедупликация входящих вебхуков на основе заголовка сообщения eventId

В приведенном выше примере мы отправили команду с метаданными. Все метаданные (например, executorId) доступны в обработчиках сообщений через атрибут Header.

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

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

Восстановление после синхронных ошибок

Одной из характеристик реактивной системы является ее отказоустойчивость.

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

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

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

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

<?php

final class MessagingConfiguration
{
    #[ServiceContext]
    public function retryStrategy()
    {
        return InstantRetryConfiguration::createWithDefaults()
                 ->withCommandBusRetry(
                      true, // is enabled
                      3, // максимальное количество попыток
                      [DatabaseConnectionFailure::class, OptimisticLockingException::class] // список исключений, при которых нужно будет запускать повторные попытки. Оставьте пустым, если нужно делать повторные попытки при всех видах исключений
                 )
    }
}

Конфигурация с повторными попытками для шины команд

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

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

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

Восстановление после асинхронных ошибок

Что касается наших асинхронных обработчиков событий. то мы также можем повторить и их.

Однако в этом случае у нас немного больше вариантов, как мы можем это сделать.

Мгновенные повторные попытки

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

<?php

final class MessagingConfiguration
{
    #[ServiceContext]
    public function retryStrategy()
    {
        return InstantRetryConfiguration::createWithDefaults()
                 ->withAsynchronousEndpointsRetry(
                      true, // is enabled
                      3, // максимальное количество попыток
                      [DatabaseConnectionFailure::class, OptimisticLockingException::class] // список исключений, при которых нужно будет запускать повторные попытки. Оставьте пустым, если нужно делать повторные попытки при всех видах исключений
                 )
    }
}

Конфигурация с повторными попытками для асинхронных обработчиков

Отложенные повторные попытки

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

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

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

<?php

#[ServiceContext]
public function errorConfiguration()
{
    return ErrorHandlerConfiguration::createWithDeadLetterChannel(
        "errorChannel",
        RetryTemplateBuilder::exponentialBackoff(1000, 10)
            ->maxRetryAttempts(3),
        "dlqChannel"
    );
}

#[ServiceActivator("dlqChannel")]
public function handle(ErrorMessage $errorMessage): void
{
    // делаем что-то с ErrorMessage, так как мы не можем восстановиться
}

Конфигурация отложенных повторных попыток

В этой конфигурации мы настроили стратегию повторных попыток, которые будут выполняться три раза. Начальная задержка составляет 1 секунду (1000 мс) и будет умножаться на 10 при каждой попытке. Это все работает из коробки, если ваш канал сообщений поддерживает задержки (SQS, RabbitMQ, Dbal, Redis).

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

Все, что передается по сети, – ненадежно, и отказоустойчивые системы проектируются с учетом этого.

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

При хорошем фундаменте система прекрасно самовосстанавливается, и вам не нужно беспокоиться о том, что внешний сервис может отключиться на пару минут. В конце концов, все будет хорошо.

Обработка неисправимых ошибок

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

Ошибка неисправима, и с ней нужно обращаться по-другому.

Ecotone решает эту проблему, внедряя канал ошибок (Error Channel).

После исчерпания повторных попыток сообщение перемещается в канал ошибок.

Канал ошибок будет работать как издатель-подписчик (если не указано иное). Это означает, что можно подключить столько обработчиков событий, сколько необходимо, чтобы обеспечить пользовательскую логику, такую ​​как отправка уведомлений в Slack или электронной почты, когда сообщение не может быть обработано.

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

Сообщение перемещается в канал ошибок, так как мы подключили Dbal Dead Letter (готовое решение Ecotone), и сохраняется в базе данных.

Затем мы разбираемся, что произошло, может это новая ошибка, просочившаяся в наш шаблон уведомления. Затем мы устраняем ошибку, релизим изменение и воспроизводим сообщение.

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

Подробнее об обработке сообщений об ошибках читайте в этом разделе документации Ecotone.

Работа с несколькими приложениями

Решение по умолчанию, которое предоставляет Dbal Dead Letter, заключается в управлении сообщениями об ошибках через командную строку.

Symfony: bin/console ecotone:deadletter:replay {messageId}
Laravel: artisan ecotone:deadletter:replay {messageId}
Ecotone Lite: $messagingSystem->runConsoleCommand("ecotone:deadletter:replay", ["messageId" => $messageId]);

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

Ecotone предоставляет Ecotone Pulse — приложение, которое объединяет сообщения об ошибках из различных сервисов и предоставляет пользовательский интерфейс для их просмотра, воспроизведения и удаления.

Подробнее об Ecotone Pulse вы можете почитать в документации Ecotone.

Заключение

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

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

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

Ecotone можно использовать с Laravel и Symfony, а его интеграция тривиальнее некуда.

Если вы не используете эти фреймворки, вам может подойти Ecotone Lite.

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

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

Демо проект интеграции доступен в этом репозитории.

В завершение материала рекомендуем бесплатный открытый урок «Браузерные тесты с Laravel Dusk». На этом уроке участники рассмотрят, как Dusk помогает легко и быстро писать браузерные End-To-End тесты, чтобы тестировать не только взаимодействие с API, но так же с реальными веб-страницами в реальном браузере.

• Записаться на открытый урок

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