Привет! Это моя первая статья на Хабре, поэтому если будут какие‑то недочёты, не судите строго)

Не так давно я разрабатывал микросервисное приложение для анализа предпочтений пользователей на основе их книг. Оно должно было обрабатывать тысячи событий в день, чтобы рекомендовать пользователям новые произведения. И в один момент я столкнулся с классической болью распределенных систем: если брокер сообщений (например, Kafka) был временно недоступен, мои события терялись навсегда.

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

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

Картина маслом — найдите 10 отличий

Предварительно я решил поискать готовые решения. На моё удивление, они даже нашлись — тот же Axon Framework неплохо покрывал Outbox. Но я не нашёл в нём реализацию для Inbox (может плохо искал, но по открытым источникам ничего найти не вышло). Более того, основная задача этого фрейма — реализовать все процессы через CQRS и Event Sourcing, что для моего решения было избыточным. Поискав ещё, я не нашёл того, что бы смогло удовлетворить мои потребности, а потому я пошёл по пути написания своего стартера.

Есть две цели, к которым я стремился, пока проектировал стартер:

1) Сделать его расширяемым настолько, насколько это возможно. Хотел я этого достичь за счёт использования уже готовых паттернов (Стратегия, Фабрика), что значительно облегчило жизнь;

2) Скрыть настолько много логики, насколько это возможно. Не везде это, правда, получилось достичь — некоторые трудности возникли с Inbox и немного с Outbox. Однако основная рутина была убрана под капот.

Давайте посмотрим, какие аннотации в нём есть. Не волнуйтесь — их всего 4.

• @EnableInboxing — аннотация, указывающая, нужно ли включать Inbox‑часть стартера или нет;

• @EnableOutboxing — аналогично предыдущей аннотации, но уже для Outbox;

• @OutboxEntity — аннотация, указывающая, что содержимое класса, над которым она навешена, является содержимым для Outbox сообщений (payload);

• @InboxListener — аннотация, которая навешивается над методом и указывает, что этот метод является обработчиком определённой очереди.

Кажется, выглядит нестрашно, верно? Если вы так подумали, то поняли всё верно — это действительно просто. И в разделе с практикой вы в этом убедитесь. Но сначала углубимся в устройство стартера чуть поподробнее.

Как по мне, логичнее всего будет описать каждую из групп компонентов отдельно. Мы пройдёмся по структуре объектов Outbox, Inbox, DLQ и Broker. Пойдём по порядку.

Компоненты Outbox

Из всего, что здесь представлено, единственное, что нужно разработчику для взаимодействия с Outbox компонентами — это OutboxContextManager. Он создаётся автоматически стартером, когда флаг outbox.enabled установлен в true. У него есть всего 2 метода — save (T message) и его перегруженный вариант save(T message, String topic). С их помощью вы добавляете сообщение в контекст Outbox (во втором случае явно указываете топик или очередь, куда отправляется сообщение; при вызове первого метода он подхватывает эти данные из @OutboxEntity). Затем шедулер подхватывает все необработанные сообщения, после чего проходится по каждому из них вызывает транзакционный метод из OutboxService для их обработки. В свою очередь OutboxService делегирует отправку компонентам Broker (об этом ниже).

Пара важных моментов:

• если deduplication‑enabled = true, шедулер выбирает только последние версии сообщений с одинаковым messageId;

• при ошибке отправки: retryCount++, сохраняется errorMessage; если retryCount >= maxRetries: сообщение удаляется из outbox и сохраняется в DLQ;

• OutboxContextManager.save() проверяет активную транзакцию через TransactionSynchronizationManager; если транзакция неактивна — пишется предупреждение в лог.

Теперь перейдём к архитектуре компонентов Inbox. Схема их связей представлена ниже:

Компоненты Inbox

Когда вы добавляете аннотацию @InboxListener над методом, бин InboxListenerRegistrar (наследник BeanPostProcessor) регистрирует метод, который обрабатывает поступающие в топики/очереди сообщения. Затем каждый бин InboxListenerRegistar попадает в InboxListenerRegistry, где для каждой очереди хранится список из методов для этой очереди (это сделано на случай, если в одну очередь будут попадать разные типы сообщений). Затем, когда в топике/очереди появляется сообщение, InboxProcessor с помощью BrokerContext читает его и сохраняет в таблицу. Затем шедулер подхватывает все необработанные сообщения и отправляет их в InboxListenerRegistry, тот берёт метод, указанный разработчиком, и обрабатывает его.

Здесь также реализованы механизм дедупликации и обработки ошибок по аналогии с Outbox. Перейдём к DLQ.

Компоненты DLQ

Когда сообщения достигают лимита по max‑retries в Outbox или Inbox, оно сохраняется в DeadLettersRepository. Затем шедулер достаёт из таблицы неопубликованные «мёртвые» сообщения, и отправляет их в топик/очередь, указанную в настройках.

А теперь перейдём к моему любимому — к сегменту Brokers.

Компоненты Brokers

Есть общий для всех брокеров интерфейс — BrokerStrategy. Он содержит 3 метода — publish(OutboxMessage message), consume(String queueName, Consumer<MessageEnvelope> handler) и publishDeadLetter(DeadLettersEntity message), что покрывает все необходимые действия для работы с брокерами. Стоит немного рассказать о методе consume. Одним из его параметров является интерфейс Consumer<MessageEnvelope>. MessageEnvelope — это фактически payload сообщения + заголовки, указываемые при отправке сообщения самим стартером. Важно отметить — пока эти заголовки на настраиваемые.

Для интерфейса BrokerStrategy уже готовы две реализации — KafkaBrokerStrategy и RabbitBrokerStrategy. Однако просто так они недоступны. За получение конкретной реализации отвечает BrokerStrategyFactory. У него есть метод getStrategy(Brokers brokerType), который по типу брокера, указанного в iobox.broker берёт нужную реализацию и передаёт её в BrokerContext. BrokerContext ничего не знает о конкретной реализации и работает только с интерфейсом BrokerStrategy.

Теперь, когда мы описали архитектуру в общем, можем переходить к практике.

Для демонстрации работы стартера я создам простое REST‑приложение со структурой, приведённой ниже. Для простоты мы сделаем так, что это приложение будет и отправлять, и читать свои же сообщения.

Структура приложения для проверки работы стартера

Предварительно надо скачать стартер и добавить зависимость в проект. Т.к. я использую Maven (любители Gradle, не бейте), то в pom.xml надо написать в dependencies это:

<dependency>
    <groupId>org.jedi_bachelor</groupId>
    <artifactId>spring-iobox-starter</artifactId>
    <version>1.0.1-alpha</version>
</dependency>

А также напишем application.yaml. Скажу сразу — я буду приводить его по порядку, потому что информации слишком много. Начнём с Outbox:

outbox:
   enabled: true
   max-retries: 5
   deduplication-enabled: true
   retention-days: 7
   scheduler:
     enabled: true
     interval: 5000
     cleanup-cron: "0 0 3 * * *"
   context-manager:
     enabled: true
     default-topic: "default"

Ниже приведу описание каждого из полей:

• outbox.enabled — флаг, включающий или выключающий поддержку готового outbox;

• outbox.max‑retries — максимальное количество попыток обработать входящее сообщение. Если это число достигается, сообщение удаляется (если DLQ не поддерживается) или отправляется в DLQ;

• outbox.deduplication‑enabled — поддерживается ли дедупликация в Outbox или нет;

• outbox.retention‑days — количество дней, которые сообщения могут находиться в таблице. Если сообщения лежат дольше этого срока, они удаляются;

• outbox.scheduler.enabled — поддерживается ли готовый шедулер или нет;

• outbox.scheduler.interval — период (в миллисекундах), через которое происходит новый обход шедулера;

• outbox.scheduler.cleanup‑cron — CRON‑задача, описывающая порядок, когда будет очищаться база данных с Outbox;

• outbox.content‑manager.enabled — поддерживается ли встроенный OutboxContentManager или нет;

• outbox.content‑manager.default‑topic — название топика (очереди) по умолчанию.

Думаю, достаточно простая и понятная настройка. Теперь, когда мы ознакомились с Outbox, перейдём к рассмотрению Inbox:

inbox:
   enabled: true
   max-retries: 5
   deduplication-enabled: true
   retention-days: 7
   scheduler:
     enabled: true
     interval: 5000
     cleanup-cron: "0 0 3 * * *"
 
   kafka:
     topics: "user-created,order-created,test-success"

Как видите, структура во многом повторяет Outbox. Единственное сильное отличие — inbox.kafka.topics. В этом поле указывается через запятую названия топиков, которые стартер будет прослушивать. Это сделано для того, чтобы Kafka автоматически создавала топики, если их нет (мы будем использовать только «user‑created»; остальное приведено как демонстрация, как надо писать, если топиков несколько).

Теперь перейдём к DLQ (Dead Letters Queue):

dlq:
   enabled: true
   dlqName: "dead-letters"
   scheduler:
     enabled: true
     interval: 5000

Во многом ситуациях похожая. dlq.dlqName — название очереди для отправки сообщений с «мёртвыми» письмами. По умолчанию принимает значение «dead‑letters», но вы всегда можете его поменять.

Ещё один важный момент — так как стартер поддерживает интеграцию как с Kafka, так и с RabbitMQ, вы можете явно указать, что именно будете использовать. Пока доступны только «kafka» и «rabbit». По умолчанию стоит значение «kafka». Можете поменять на «rabbit», как показано ниже:

iobox:
  broker: rabbit

Остался последний штрих — накидать настройки для Kafka:

spring:
  jpa:
    hibernate:
      ddl-auto: update
  kafka:
    bootstrap-servers: localhost:9093
    consumer:
      group-id: bs-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "*"
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer

Предварительно поднимите ещё сервер Kafka (если хотите — Rabbit). Для просмотра сообщений я буду использовать Kafka UI — можете его также запустить в Docker.

Тройка из Kafka + Zookeper + Kafka UI

Главный класс выглядит очень просто:

import org.jedi_bachelor.ioboxstarter.annotations.EnableInboxing;
import org.jedi_bachelor.ioboxstarter.annotations.EnableOutboxing;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
@EnableOutboxing
@EnableInboxing
public class IoboxtestApplication {
    public static void main(String[] args) {
       SpringApplication.run(IoboxtestApplication.class, args);
    }
}

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

import lombok.Getter;
import lombok.RequiredArgsConstructor;
import org.jedi_bachelor.ioboxstarter.annotations.OutboxEntity;

@OutboxEntity(queueName = "user-created")
@Getter
@RequiredArgsConstructor
public class TestEvent {
    private final Long id;
    private final String message;
    private final boolean shouldFail;
}

Здесь в аннотации @OutboxEntity обязательно надо указать, в какую очередь должно отправляться сообщение. Теперь реализуем класс TestController, где создадим два эндпоинта:

• POST /success/{id} — отправка успешного сообщения (в ходе обработки должны появиться сообщения в таблицах Inbox и Outbox);

• POST /fail/{id} — отправка провального сообщения (на нём мы увидим, что DLQ работает).

Текст класса представлен ниже:

import com.example.ioboxtest.service.TestService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/test")
@RequiredArgsConstructor
@Slf4j
public class TestController {
    private final TestService testService;

    @PostMapping("/success/{id}")
    public ResponseEntity<String> sendSuccess(@PathVariable Long id, @RequestParam String message) {
        this.testService.sendSuccessEvent(id, message);

        return ResponseEntity.ok("Success event sent with id: " + id);
    }

    @PostMapping("/fail/{id}")
    public ResponseEntity<String> sendFail(@PathVariable Long id, @RequestParam String message) {
        this.testService.sendFailEvent(id, message);

        return ResponseEntity.ok("Fail event sent with id: " + id);
    }
}

Теперь добавим класс TestService:

import com.example.ioboxtest.entity.TestEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.jedi_bachelor.ioboxstarter.OutboxContextManager;
import org.springframework.stereotype.Service;

@Service
@RequiredArgsConstructor
@Slf4j
public class TestService {
    private final OutboxContextManager outboxContextManager;

    public void sendSuccessEvent(Long id, String message) {
        TestEvent event = new TestEvent(id, message, false);

        this.outboxContextManager.save(event);

        log.info("Sent success event: {}", message);
    }

    public void sendFailEvent(Long id, String message) {
        TestEvent event = new TestEvent(id, message, true);

        this.outboxContextManager.save(event);

        log.info("Sent fail event: {}", message);
    }
}

ПРИМЕЧАНИЕ: ранее я отмечал, что компоненты Outbox следят за тем, выполняете ли вы обработку Outbox‑сообщений в транзакционных методах или нет. Запустив эти примеры, вы увидите соответствующие логи с предупреждением, но в будущем лучше просто навесить @Transactional там, где это надо;)

Последнее, что нам надо сделать — это написать обработчик для Inbox сообщений. Давайте реализуем его, создав класс TestHandler:

import com.example.ioboxtest.entity.TestEvent;
import lombok.extern.slf4j.Slf4j;
import org.jedi_bachelor.ioboxstarter.annotations.InboxListener;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class TestHandler {
    @InboxListener(queueName = "user-created")
    public void handle(TestEvent event) {
        log.info("Message: Processing event with id: {}, message: {}",
                event.getId(), event.getMessage());

        if (event.isShouldFail()) {
            throw new RuntimeException("Test exception - message should go to DLQ");
        }

        log.info("SUCCESS: Event processed successfully");
    }
}

Здесь есть один обработчик — для очереди «user‑created». При получении сообщения мы будем просто выводить о нём информацию. Если флаг сообщения shouldFail установлен в true, мы выкинем исключение.

Теперь перейдём к тесту. Когда вы запустите приложение, то увидите, что Iobox создаёт топик «user‑created». Отправим запрос с созданием успешного сообщения:

Отправили запрос с корректным сообщением

Логи обработки запроса

Как видим из логов — запрос был успешно отправлен, его тело помещено в Outbox таблицу, после чего шедулер находит это сообщение и обрабатывает его. Затем это же приложение видит в топике «user‑created» новое сообщение, сохраняет в Inbox, и потом шедулер обрабатывает его. Если мы залезем в Kafka UI, то действительно найдём там сообщение:

Подтверждение нахождения сообщения в очереди

Супер! А теперь давайте посмотрим обработку негативного сценария. Как вы ранее видели, мы симулировали намеренную ошибку, но этот пример покажет, что всё и правда работает. Отправим новый запрос:

Отправка запроса с некорректным сообщением

В логах после этого вы увидите всё то же самое, а затем вылетит 5 сообщений об исключении — кол‑во равно значению max‑retries. После этого в логах вы увидите это:

Логи работы DLQ

Сообщение, которое мы ранее отправили, было переслано в топик «dead‑letters». Проверим это, и… да, там оно действительно есть:

«Умершее» сообщение в DLQ

Таким образом, мы видим, что всё работает успешно)

Получение актуальной информации из баз данных

Остаётся ещё один вопрос — можно ли проверить состояние и значения сообщений, помимо просмотра топика? Для того, чтобы быстро получать актуальную информацию баз данных для Inbox, Outbox и Dead Letters, я создал соответствующие эндпоинты:

• GET /outbox — выводит все сообщения из таблицы сообщений Outbox;

• GET /inbox — аналогично как в /outbox, только для таблицы Inbox;

• GET /dead‑letters — аналогично, только для таблицы dead letters.

Проверим их, чтобы не быть голословными!

Сообщения Outbox

Сообщения в Inbox

Сообщение в DLQ

Мы получили картину, которую и ожидали — два сообщения Outbox были отправлены, но одно из них попало в Inbox, потому что имело корректный формат, а второе отправлено в dead‑letters, потому что система не смогла его обработать. Теперь у вас есть возможность получать подробную информацию о данных в таблицах ещё и таким способом.

Конечно, это только первая версия. Но уже сейчас стартер экономит часы разработки и гарантирует, что ваши события не потеряются. В планах добавить интеграцию с NoSQL базами данных (MongoDB и Cassandra), реактивным программированием и возможность отправки запросов не только через очереди или сессии, а также через REST‑запросы. Также уже готовлюсь добавлять метрики Micrometer, а вместе с тем намерен переработать настройки в application.yaml (пока они выглядят достаточно разрозненно).

Сам стартер лежит по этой ссылке. Я продолжу развивать проект и буду рад вашим звездам на GitHub, issue и pull request'ам.

Спасибо за внимание!

P. S. Статью я писал в тот момент, когда только перерабатывал интеграцию с брокерами посредством паттерна Стратегия (раньше всё было завязано на интерфейсах OutboxProducer и InboxConsumer). К моему удивлению, после изменений стартера в тестовом приложении, приведённом в статье, не поменялось ни строчки кода. Для меня это был шок, хотя, скорее всего, это что‑то из разряда нормы.