Apache ActiveMQ

Apache ActiveMQ представляет собой брокер сообщений, реализующий модель обмена сообщениями между распределёнными системами. В контексте STOMP.js он выступает серверной стороной, принимающей STOMP-протокол поверх TCP или WebSocket и обеспечивающей маршрутизацию сообщений между продюсерами и потребителями.

ActiveMQ поддерживает несколько протоколов, среди которых STOMP занимает важное место благодаря своей простоте и текстовому формату кадров (frames). Это делает его удобным для интеграции с JavaScript-клиентами, работающими в браузере или Node.js через STOMP.js.

Архитектура взаимодействия STOMP.js и ActiveMQ

Взаимодействие строится на модели клиент–брокер:

  • STOMP.js формирует текстовые STOMP-кадры
  • Брокер ActiveMQ принимает кадры через STOMP-коннектор
  • Сообщения маршрутизируются в очереди (queue) или топики (topic)
  • Подписчики получают сообщения по установленным подпискам

ActiveMQ в этой модели выполняет роль центра маршрутизации, а STOMP.js — лёгкого клиента, не требующего сложных бинарных протоколов.

STOMP-коннектор в ActiveMQ

Для работы STOMP.js с ActiveMQ требуется включённый STOMP-коннектор. В конфигурации ActiveMQ он задаётся через activemq.xml:

<transportConnectors>
    <transportConnector name="stomp" uri="stomp://0.0.0.0:61613"/>
</transportConnectors>

Порт 61613 является стандартным для STOMP поверх TCP.

Для WebSocket-сценариев используется дополнительная настройка:

<transportConnector name="stomp+ws" uri="ws://0.0.0.0:61614"/>

Это позволяет браузерному STOMP.js подключаться напрямую без промежуточных прокси.

Подключение STOMP.js к ActiveMQ

Клиент STOMP.js создаёт соединение с брокером через WebSocket:

import { Client } from '@stomp/stompjs';

const client = new Client({
  brokerURL: 'ws://localhost:61614',
  reconnectDelay: 5000,
  heartbeatIncoming: 4000,
  heartbeatOutgoing: 4000
});

client.onConn ect = (frame) => {
  console.log('Connected to ActiveMQ');

  client.subscribe('/queue/orders', (message) => {
    const body = JSON.parse(message.body);
    console.log('Received:', body);
  });
};

client.activate();

ActiveMQ интерпретирует адрес /queue/orders как очередь orders.

Очереди и топики в ActiveMQ

ActiveMQ поддерживает две базовые модели доставки сообщений:

Очереди (Queue)

Очередь обеспечивает модель point-to-point:

  • каждое сообщение доставляется одному потребителю
  • используется для задач обработки, где важна уникальная обработка
  • STOMP-адресация: /queue/name

Пример подписки:

client.subscribe('/queue/taskQueue', callback);

Топики (Topic)

Топик реализует pub/sub модель:

  • каждое сообщение доставляется всем подписчикам
  • используется для событий и рассылок
  • STOMP-адресация: /topic/name

Пример:

client.subscribe('/topic/updates', callback);

ActiveMQ автоматически разделяет маршрутизацию по префиксам /queue и /topic.

Отправка сообщений в ActiveMQ через STOMP.js

Отправка сообщений выполняется через метод publish (или send в старых версиях):

client.publish({
  destination: '/queue/orders',
  body: JSON.stringify({
    id: 123,
    status: 'created'
  })
});

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

Формат STOMP-кадров в ActiveMQ

STOMP протокол основан на текстовых кадрах, содержащих:

  • команду (SEND, SUBSCRIBE, CONNECT)
  • заголовки
  • тело сообщения

Пример SEND-кадра:

SEND
destination:/queue/orders
content-type:application/json

{"id":123,"status":"created"}

ActiveMQ парсит кадр и направляет сообщение в соответствующий destination.

Поддержка постоянных подписок

ActiveMQ поддерживает durable subscriptions для топиков. Это позволяет сохранять сообщения для офлайн-подписчиков.

Настройка на стороне клиента:

client.subscribe('/topic/events', callback, {
  id: 'unique-subscription-id',
  persistent: 'true'
});

На стороне ActiveMQ подписка идентифицируется по clientId и subscriptionName.

Подтверждение сообщений (ACK)

ActiveMQ поддерживает разные режимы подтверждения:

  • auto
  • client
  • client-individual

В STOMP.js:

client.subscribe('/queue/orders', (message) => {
  const data = JSON.parse(message.body);

  message.ack();
}, { ack: 'client' });

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

Транзакционность сообщений

ActiveMQ поддерживает транзакционные сессии STOMP:

client.begin('tx1');

client.publish({
  destination: '/queue/orders',
  body: 'data'
});

client.commit('tx1');

При ошибке можно выполнить:

client.abort('tx1');

Это обеспечивает атомарность группы сообщений.

Heartbeat и стабильность соединения

ActiveMQ и STOMP.js используют heartbeat для контроля живости соединения.

Конфигурация клиента:

heartbeatIncoming: 10000,
heartbeatOutgoing: 10000

ActiveMQ отвечает на heartbeat кадрами \n, предотвращая разрыв соединения NAT или прокси.

Фильтрация сообщений через селекторы

ActiveMQ поддерживает message selectors на основе SQL92:

client.subscribe('/queue/orders', callback, {
  selector: "priority = 'high'"
});

Со стороны ActiveMQ фильтрация происходит до доставки сообщения клиенту.

Persistent и non-persistent сообщения

ActiveMQ различает типы доставки:

  • persistent — сохраняются на диск
  • non-persistent — живут в памяти

STOMP.js задаёт это через заголовки:

client.publish({
  destination: '/queue/orders',
  headers: {
    persistent: 'true'
  },
  body: 'message'
});

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

Управление соединениями ActiveMQ

ActiveMQ отслеживает соединения STOMP-клиентов:

  • уникальный client-id
  • список активных подписок
  • состояние heartbeat
  • транзакции

При разрыве соединения брокер может сохранить состояние durable подписок и восстановить доставку после реконнекта клиента.

Ошибки и диагностика STOMP-взаимодействия

ActiveMQ отправляет ERROR кадры при проблемах:

ERROR
message:Subscription failed
content-type:text/plain

Subscription not authorized

На стороне STOMP.js обработка:

client.onStompEr ror = (frame) => {
  console.error(frame.headers['message']);
};

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

Масштабирование ActiveMQ при работе со STOMP.js

ActiveMQ может работать в нескольких режимах:

  • standalone broker
  • master/slave
  • network of brokers

При STOMP.js нагрузке масштабирование достигается через:

  • распределение очередей
  • network connectors
  • балансировку WebSocket-коннектов

STOMP остаётся единым протоколом на уровне клиента, независимо от внутренней архитектуры брокера.

Особенности работы в браузере

STOMP.js в браузере через ActiveMQ требует:

  • WebSocket-коннектор ActiveMQ
  • корректной CORS/WS конфигурации
  • отсутствия прямого TCP доступа (ограничение браузера)

Типичный endpoint:

ws://host:61614

При этом брокер остаётся полностью прозрачным для клиента, который оперирует только destination-адресами.

Оглавление