RabbitMQ

RabbitMQ — внутренний транспорт сообщений DataHUB. Брокер обслуживает очереди обмена между подключёнными системами и backend'ом, а также управляющий и нотификационный трафик. Поставляется как отдельный Docker Compose блок и не требует развертывания внешнего брокера.

Эта статья описывает конфигурацию блока rabbitmq и принципы его взаимодействия с Сервером DataHUB. Установка по шагам — в Быстром старте. Логика очередей и автоматическая конфигурация exchange/queue — в Архитектуре / Работа с очередями.

Версия и образ

Образ поставляется вендором и собран на базе официального rabbitmq:3-management (RabbitMQ 3.x + плагин Management UI). В образ добавлен HEALTHCHECK через REST API брокера:

FROM rabbitmq:3-management
RUN apt-get update && apt-get install -y curl
HEALTHCHECK --interval=30s --timeout=10s --start-period=10s --retries=3 \
    CMD curl -f -u $RABBITMQ_DEFAULT_USER:$RABBITMQ_DEFAULT_PASS http://localhost:15672/api/healthchecks/node || exit 1

Плагин Management UI включён по умолчанию — он используется и для healthcheck, и для веб-интерфейса администратора, и для серверной автоматической конфигурации (Сервер DataHUB создаёт exchange и очереди через тот же Management API).

Структура блока

Каталог блока — /etc/datahub/rabbitmq/:

/etc/datahub/rabbitmq/
├── docker-compose.yml   # описание контейнера
├── .env.template        # шаблон переменных окружения
├── .env                 # реальные значения (создаётся при установке)
└── data/                # данные брокера (bind-mount в контейнер)

Файл docker-compose.yml (фактическое содержимое поставки):

services:
  rabbitmq:
    image: registry.gitlab.com/datahub/v3/services/rabbitmq:dev
    container_name: rabbitmq
    hostname: rabbitmq
    volumes:
      - ./data:/var/lib/rabbitmq
    expose:
      - 5672   # AMQP broker
      - 15672  # web management
    env_file:
      - .env
    restart: always
    networks:
      - dh_network

Ключевые особенности:

expose без ports — оба порта (5672 AMQP и 15672 Management UI) видны только внутри Docker-сети dh_network. Снаружи доступ к брокеру не выставляется.
./data:/var/lib/rabbitmq — состояние брокера (определения exchange/queue, метаданные пользователей, persistent-сообщения) хранится на хост-файловой системе и переживает пересоздание контейнера.
hostname: rabbitmq — DNS-имя внутри сети, по которому к брокеру обращается backend.

Конфигурация (`.env`)

Шаблон .env.template:

RABBITMQ_DEFAULT_USER=admin
RABBITMQ_DEFAULT_PASS={{RABBITMQ_PASSWORD}}
RABBITMQ_DEFAULT_VHOST=/

Переменная	Назначение	Значение по умолчанию
`RABBITMQ_DEFAULT_USER`	Имя суперпользователя брокера, создаётся при первом старте	`admin`
`RABBITMQ_DEFAULT_PASS`	Пароль суперпользователя	задаётся вручную, минимум 16 символов
`RABBITMQ_DEFAULT_VHOST`	Виртуальный хост по умолчанию	`/`

Те же креды затем подставляются в .env блока backend (RABBITMQ_USER, RABBITMQ_PASSWORD) — backend подключается как admin для подписки на очереди и автоматической конфигурации топологии. Изменение пароля выполняется согласованно в обоих блоках с последующим перезапуском backend.

Генерация криптостойкого пароля:

head -c 32 /dev/urandom | base64

Каталог данных

Каталог ./data соответствует /var/lib/rabbitmq внутри контейнера. Содержит:

mnesia/ — встроенная база метаданных брокера (определения exchange, queue, binding, пользователи, права).
журнал персистентных сообщений — для очередей с durable: true и сообщений с delivery_mode: 2.

Все эти данные переживают рестарт и обновление контейнера. При полной переустановке (включая смену имени узла) каталог можно очистить — топология будет восстановлена backend'ом автоматически при следующем старте.

Автоматическая конфигурация топологии

DataHUB управляет топологией RabbitMQ программно. После старта Сервера DataHUB он:

Подключается к Management API по адресу http://rabbitmq:15672 (переменная RABBITMQ_MANAGEMENT_URL).
Проверяет наличие предопределённых exchange (amq.direct, amq.fanout, amq.topic, amq.headers, amq.match, amq.rabbitmq.trace) — они создаются самим брокером.
Создаёт пользовательские exchange и очереди по конфигурации в БД (типы обмена, маршрутизация, подключённые клиентские системы).
Периодически (раз в 60 секунд по умолчанию) синхронизирует состояние: если кто-то удалил очередь снаружи — она будет пересоздана.

Параметры из application.yml backend'а:

datahub:
  exchange:
    rabbitmq:
      default-exchange-type: topic
      predefined-v-host: /
      sync-interval-ms: 60000          # периодичность сверки топологии
      client-template-pool:
        cleanup-interval: 900000       # очистка неиспользуемых шаблонов раз в 15 мин
        cleanup-threshold: 300000      # порог неактивности — 5 мин
        max-in-flight-messages: 5      # одновременно обрабатываемых сообщений на шаблон
        ack-ttl-seconds: 120           # TTL подтверждения — после возврат в очередь
      connection:
        timeout: 5000                  # таймаут подключения
        heartbeat: 30                  # heartbeat-интервал в секундах
        network-recovery-interval: 5000
        channel-rpc-timeout: 5000
        automatic-recovery-enabled: true
        topology-recovery-enabled: false  # топология восстанавливается backend'ом, не клиентом

topology-recovery-enabled: false намеренно — восстановление exchange/queue выполняется именно backend'ом через Management API, а не клиентским AMQP-клиентом. Это исключает гонки при одновременном восстановлении нескольких компонентов.

Подробности логики синхронизации топологии — в Архитектуре / Работа с очередями.

Подключение Сервера DataHUB

Backend подключается к брокеру по AMQP через сервисное имя rabbitmq в общей Docker-сети. Параметры подписчика очередей (application.yml):

spring:
  rabbitmq:
    host: ${RABBITMQ_HOST}        # rabbitmq
    port: ${RABBITMQ_PORT}        # 5672
    username: ${RABBITMQ_USER}
    password: ${RABBITMQ_PASSWORD}
    virtual-host: ${RABBITMQ_VHOST}   # /
    listener:
      simple:
        acknowledge-mode: auto
        retry:
          enabled: true
          initial-interval: 2000
          max-attempts: 3
          multiplier: 2.0

При нештатной обработке сообщения Spring AMQP делает до трёх попыток с экспоненциальной задержкой (2s, 4s, 8s). Логика повторной обработки сообщений после исчерпания попыток описана в Концепции / Жизненный цикл сообщения.

Healthcheck и состояние

Healthcheck опрашивает /api/healthchecks/node через curl каждые 30 секунд. Проверка статуса контейнера:

sudo docker ps --filter name=rabbitmq --format "table {{.Names}}\t{{.Status}}"

В колонке STATUS должно быть Up <время> (healthy). Состояние unhealthy указывает на проблемы — типичные причины: неверный пароль администратора, повреждённая mnesia (/var/lib/rabbitmq/mnesia), переполнение диска (RabbitMQ блокирует приём сообщений при заполнении диска ниже порога).

Журнал контейнера:

sudo docker logs -f --tail 200 rabbitmq

Управление через Management UI

Прямого доступа к Management UI (15672) снаружи нет — порт expose-ится только в dh_network. Веб-интерфейс публикуется через nginx по схеме reverse-proxy на отдельном порту хоста.

При схеме развертывания «два домена» (по умолчанию) доступ к Management UI:

https://lk.<your-domain>:15672/

Конфиг прокси лежит в /etc/datahub/nginx/config/conf.d/rmq-admin.conf и проксирует все запросы на http://rabbitmq:15672 с поддержкой WebSocket (используется живым UI для обновления статистики). Аутентификация — через тот же admin из .env блока rabbitmq.

Через Management UI выполняются типовые операции диагностики:

просмотр очередей, exchange и привязок;
ручной просмотр и удаление сообщений из конкретной очереди;
ручная отправка тестового сообщения;
мониторинг RPS, размера очередей, числа коннекшенов и каналов;
управление пользователями и виртуальными хостами (для DataHUB обычно не требуется).

CLI-доступ (rabbitmqctl)

Команды управления через CLI выполняются внутри контейнера. Общая информация о состоянии узла:

sudo docker exec -it rabbitmq rabbitmqctl status

Список очередей с числом сообщений и подписчиков:

sudo docker exec -it rabbitmq rabbitmqctl list_queues name messages consumers

Список exchange:

sudo docker exec -it rabbitmq rabbitmqctl list_exchanges name type

Полный справочник команд — rabbitmqctl --help или официальная документация RabbitMQ.

Резервное копирование

Состояние брокера (определения exchange/queue/binding/user) экспортируется в JSON через rabbitmqctl:

sudo docker exec rabbitmq rabbitmqctl export_definitions /var/lib/rabbitmq/definitions.json

Извлечение дампа из контейнера на хост:

sudo docker cp rabbitmq:/var/lib/rabbitmq/definitions.json /var/backups/datahub/rabbitmq-definitions-$(date +%F).json

Импорт обратно (например, после переустановки):

sudo docker cp /var/backups/datahub/rabbitmq-definitions-YYYY-MM-DD.json rabbitmq:/var/lib/rabbitmq/definitions.json
sudo docker exec rabbitmq rabbitmqctl import_definitions /var/lib/rabbitmq/definitions.json

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

Полная стратегия бэкапа платформы — в Резервном копировании.

Что дальше

Reverse Proxy — публикация Management UI через nginx, ограничение доступа.
Архитектура / Работа с очередями — как DataHUB создаёт и обслуживает топологию.
Концепции / Жизненный цикл сообщения — как сообщения проходят через брокер.
Диагностика / Проблемы RabbitMQ — типовые ошибки и их разбор.