Настройка очередей

Модель очередей в STOMP опирается на концепцию брокера сообщений, где клиент не взаимодействует напрямую с серверной логикой, а отправляет и получает сообщения через заранее определённые маршруты доставки — destinations. В контексте STOMP.js это выражается через подписки на очереди (queues) и топики (topics), которые интерпретируются конкретным брокером (RabbitMQ, ActiveMQ, Apollo, Spring WebSocket broker и др.).

Очередь (queue) представляет собой структуру типа point-to-point: одно сообщение обрабатывается одним получателем. В отличие от топиков, где действует модель pub/sub, очередь гарантирует распределение сообщений между потребителями.

В STOMP.js работа с очередями сводится к корректной настройке destination-строк и параметров подписки.

Формирование адреса очереди (destination naming)

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

  • /queue/... — очередь
  • /topic/... — широковещательная рассылка
  • /exchange/... — маршрутизация через exchange (RabbitMQ-специфично)

Пример адреса очереди:

/queue/orders
/queue/payment.process
/queue/user.notifications

STOMP.js не интерпретирует эти строки — они полностью зависят от брокера. Однако соблюдение соглашений критично для совместимости.

Подключение клиента к брокеру

Перед работой с очередями необходимо установить соединение через WebSocket и STOMP-клиент.

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

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

Ключевые параметры:

  • brokerURL — WebSocket-эндпоинт брокера
  • reconnectDelay — автоматическое восстановление соединения
  • heartbeatIncoming / heartbeatOutgoing — контроль живости соединения

Подписка на очередь

Основной механизм получения сообщений — метод subscribe.

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

    console.log('Получено сообщение:', body);
  });
};

client.activate();

Каждое сообщение приходит в виде объекта message, где:

  • message.body — строковое представление payload
  • message.headers — метаданные
  • message.ack() — подтверждение обработки (если включён режим ACK)

Режимы подтверждения сообщений (ACK)

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

AUTO ACK

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

client.subscribe('/queue/orders', handler, { ack: 'auto' });

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

CLIENT ACK

Требует явного подтверждения:

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

  processOrder(data);

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

Если ack() не вызван, брокер может повторно доставить сообщение.

CLIENT-IN-TRANSACTION

Позволяет группировать подтверждения:

client.subscribe('/queue/orders', handler, {
  ack: 'client-individual'
});

Используется в сценариях с частичной обработкой потока.

Отправка сообщений в очередь

STOMP.js позволяет отправлять сообщения в очередь через метод publish.

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

Отправка не гарантирует обработку — гарантии зависят от брокера и конфигурации очереди.

Очереди с приоритетами

Некоторые брокеры поддерживают приоритет сообщений. В STOMP.js это реализуется через headers.

client.publish({
  destination: '/queue/orders',
  body: JSON.stringify({ orderId: 124 }),
  headers: {
    priority: '9'
  }
});

Поддержка приоритетов зависит от брокера (например, ActiveMQ поддерживает, RabbitMQ требует настройки policy).

Долгоживущие очереди и устойчивость подписок

STOMP сам по себе не хранит состояние очередей. Это задача брокера.

Однако можно использовать durable subscriptions (если поддерживается):

client.subscribe('/queue/orders', handler, {
  id: 'orders-subscription',
  persistent: 'true'
});

Важные аспекты:

  • идентификатор подписки (id) обязателен для восстановления
  • брокер должен поддерживать durable subscriptions
  • восстановление происходит после переподключения клиента

Prefetch и контроль нагрузки

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

В STOMP.js это не настраивается напрямую, но передаётся через headers:

client.subscribe('/queue/orders', handler, {
  'prefetch-count': 10
});

Типичное поведение:

  • маленький prefetch → равномерная нагрузка
  • большой prefetch → высокая пропускная способность, но риск перегрузки клиента

Обработка ошибок очередей

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

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

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

Если брокер поддерживает nack, сообщение возвращается в очередь.

Переподключение и восстановление подписок

При разрыве соединения STOMP.js автоматически восстанавливает WebSocket-соединение (если включён reconnectDelay), но подписки необходимо учитывать отдельно.

const subscriptions = [];

client.onConn ect = () => {
  subscriptions.push(
    client.subscribe('/queue/orders', handler)
  );
};

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

Очереди и масштабирование потребителей

При нескольких клиентах, подписанных на одну очередь:

  • сообщения распределяются между ними
  • каждый message доставляется только одному потребителю
  • балансировка зависит от брокера

Пример сценария:

  • service A, B, C подписаны на /queue/orders
  • брокер распределяет нагрузку
  • обеспечивается горизонтальное масштабирование обработки

Очереди в связке с транзакциями

Некоторые брокеры поддерживают транзакционные сообщения:

client.begin().then((transaction) => {
  client.publish({
    destination: '/queue/orders',
    body: JSON.stringify({ id: 1 }),
    transaction
  });

  transaction.commit();
});

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

Типовые ошибки при настройке очередей

Несовпадение destination

/queue/order ≠ /queue/orders

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

Отсутствие ACK при client mode

Если не вызвать ack(), очередь может заблокироваться повторной доставкой сообщений.

Потеря подписки при reconnect

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

Перегрузка клиента

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

Архитектурные особенности использования очередей

Очереди в STOMP.js обычно применяются в следующих сценариях:

  • обработка фоновых задач (job queues)
  • распределённая обработка заказов
  • очереди уведомлений
  • интеграция микросервисов через брокер сообщений
  • потоковая обработка событий с гарантией доставки

Поведение системы определяется не STOMP.js, а комбинацией:

  • брокера сообщений
  • политики очередей
  • стратегии подтверждений
  • архитектуры потребителей
Оглавление