Перейти к содержанию

Работа с очередями

Шинная часть Сервера DataHUB опирается на RabbitMQ как на брокер сообщений. Все взаимодействия с RMQ инкапсулированы в пакете com.datahub.exchange.rabbit.* и сервисах com.datahub.exchange.service.*. Эта статья описывает топологию очередей, схему именования, синхронизацию топологии с конфигурацией и поведение в нештатных ситуациях.

Логические очереди и их физическая реализация

С точки зрения пользователя шины (администратора, разработчика клиента) существуют логические очереди вроде «очередь заказов для торговой базы». Под капотом каждая логическая очередь представлена конкретной очередью в RabbitMQ. Имя физической очереди детерминировано: формируется из кода системы-получателя и типа сообщения, по стабильному шаблону.

Это означает:

  • По имени очереди в RabbitMQ Management UI легко определить, кому она принадлежит и какие сообщения хранит.
  • Backend не «придумывает» имена — он строит их по правилу из конфигурации в PostgreSQL.
  • Изменения имён очередей возможны только при изменениях схемы (мажорные релизы); внутри минора имена стабильны.

Имена exchange'ей и binding'ов строятся по тем же принципам.

Жизненный цикл топологии

В RabbitMQ нет единого источника истины о том, какие очереди должны существовать — этим управляет backend на основе конфигурации в PostgreSQL.

При старте backend:

  1. Backend читает из БД конфигурацию обмена (системы, маршруты, подписки).
  2. Через RabbitMQ Management API проверяет фактическое состояние брокера.
  3. Создаёт недостающие очереди, exchange'и и binding'и.
  4. Удалять лишнее не торопится — это решение, принимаемое осторожно (см. ниже).

В работе backend:

Сервис exchange.service.RabbitMQSyncService периодически (раз в N миллисекунд, конфигурируется через EXCHANGE_RABBIT_SYNC_INTERVAL_MS) сверяет состояние RMQ с конфигурацией в БД. Это нужно для случаев:

  • Кто-то вручную удалил очередь через UI/CLI RMQ — backend восстановит её.
  • В админке добавлен новый маршрут — backend создаст соответствующую очередь.
  • Перезапущен сам RMQ — backend сверится и поднимет недостающее.

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

При удалении подписки в админке:

Backend помечает соответствующую очередь как «к удалению» в БД. Удаление выполняется отложенно — после паузы, чтобы успели обработаться оставшиеся в очереди сообщения. Конкретная политика — параметр конфигурации.

Управление топологией через Management API

Прямые операции с очередями (create / delete / bind / unbind / inspect) идут через RabbitMQ Management API — HTTP-интерфейс брокера, отдельный от AMQP-протокола. Реализация — exchange.rabbit.RabbitMQManagementClient (Spring WebClient над HTTP API брокера).

Преимущества такого подхода (вместо использования AMQP-команд declare/delete):

  • Чёткое разделение «административные операции» и «операции обмена сообщениями».
  • Возможность получать метаданные очередей (длина, in-flight, потребители) — это часть Management API, недоступная через чистый AMQP.
  • Простая интеграция с healthcheck и мониторингом.

Семантика очередей

Все очереди создаются с одинаковым набором свойств:

  • Durable. Очередь переживает перезагрузку RMQ.
  • Persistent messages. Сообщения записываются на диск, не теряются при сбое брокера.
  • Manual acknowledgement. Сообщения удаляются из очереди только после явного подтверждения получателем, не сразу при выдаче.
  • No automatic TTL. Сообщения не удаляются по времени жизни — лежат, пока не будут обработаны или явно удалены администратором.

Эти свойства обеспечивают гарантии доставки at-least-once.

Ограничение in-flight

Чтобы один медленный или зависший получатель не «забрал» слишком много сообщений и не парализовал работу остальных consumer'ов, backend применяет ограничение на количество одновременно находящихся в обработке (in-flight) сообщений у одного получателя. При превышении возвращается ошибка ToManyInFlightException (см. exchange.error.ToManyInFlight*) — клиент должен сначала подтвердить или вернуть уже выданные сообщения, а потом запрашивать новые.

Конкретный лимит — параметр конфигурации Exchange-контура. Это часть защиты от ситуаций «забрал и пропал».

Сериализация сообщений

Сообщения сериализуются в JSON и кладутся в очередь как байты с content-type: application/json. Headers RabbitMQ используются для служебных полей, не дублирующихся в payload (correlation-id, идентификатор пакета, версия) — это позволяет backend'у читать заголовок без полной десериализации тела.

Кеш конфигурации

Конфигурация обмена (системы, маршруты, подписки) меняется относительно редко, а читается на каждом запросе — было бы дорого ходить за ней в БД каждый раз. Backend кеширует её в Caffeine с двойной стратегией:

  • TTL — данные кеша считаются актуальными в течение N секунд.
  • Явная инвалидация — при изменении конфигурации через LK-контур backend сбрасывает соответствующий кеш через событие.

Это позволяет Exchange-контуру работать с пиковой нагрузкой без существенной зависимости от PostgreSQL.

Очереди нотификаций

Помимо обменных очередей, в RabbitMQ живут служебные очереди backend'а — для асинхронной отправки email/SMS. Конфигурируются модулем notification.rabbitmq.NotificationRabbitMQConfig. Эти очереди отделены от обменных и обслуживаются отдельным consumer'ом внутри backend'а (notification.rabbitmq.receiver.*). Внешним клиентам они не видны.

Поведение при сбоях

Сбой Поведение
Перезагрузка backend Сервис синхронизации поднимет недостающие очереди при старте; in-flight сообщения вернутся в очередь по таймауту RMQ
Перезагрузка RabbitMQ Очереди и сообщения переживают (durable + persistent). Backend дождётся восстановления связи и переподключится. Сервис синхронизации сверит топологию
Ручное удаление очереди в RMQ UI Сервис синхронизации восстановит на следующем цикле
Потеря Mnesia (БД RMQ) Топология полностью пересоздаётся синхронизатором. Сообщения, бывшие в очередях на момент потери, теряются — отдельный сценарий восстановления, см. Резервное копирование
Сетевой обрыв между backend и RMQ Spring AMQP переподключается автоматически; in-flight сообщения вернутся в очередь и будут выданы снова

Что НЕ используется

  • Dead Letter Exchange (DLX). Сообщения, не подтверждённые получателем, возвращаются в исходную очередь (см. Концепции / Жизненный цикл сообщения), а не уходят в отдельную DLQ. Эта концепция неприменима, тк все сообщения доставляются адаптеру и вопрос что делать с данными решается там.
  • Отдельная retry-очередь с delay. Spring AMQP retry на стороне consumer'а — in-process, не через выделенную очередь.
  • Quorum queues / streams. Используются классические persistent-очереди — для текущего масштаба этого достаточно. Переход на quorum-очереди для высокой доступности рассматривается в Масштабировании.

Что дальше