Параметры подписки

Подписка в STOMP.js управляется методом subscribe(). Помимо адреса назначения и callback-функции, метод принимает объект параметров, позволяющий управлять поведением подписки на уровне STOMP-протокола и брокера сообщений.

Базовая форма подписки:

client.subscribe(destination, callback, headers);

Пример:

client.subscribe('/topic/news', message => {
    console.log(message.body);
}, {
    id: 'news-subscription'
});

Третий аргумент содержит параметры подписки, которые преобразуются в STOMP-заголовки кадра SUBSCRIBE.

Структура объекта параметров

Объект параметров представляет собой набор заголовков:

{
    id: 'sub-1',
    ack: 'client',
    durable: 'true'
}

STOMP.js не ограничивает список доступных параметров. Библиотека передаёт их брокеру без дополнительной обработки.

Это позволяет:

  • использовать стандартные STOMP-заголовки;
  • подключать специфические параметры брокеров;
  • управлять маршрутизацией;
  • настраивать подтверждение сообщений;
  • активировать долговременные подписки.

Параметр id

Назначение идентификатора подписки

Заголовок id задаёт уникальный идентификатор подписки внутри STOMP-соединения.

Пример:

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

Идентификатор используется:

  • при отписке;
  • для отслеживания подписок;
  • брокером сообщений;
  • при подтверждении сообщений;
  • в диагностике соединений.

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

Если id не указан, STOMP.js создаёт его автоматически.

client.subscribe('/topic/chat', message => {
    console.log(message.body);
});

Внутренне библиотека формирует уникальное значение:

sub-0
sub-1
sub-2

Явное указание id

Ручное задание идентификатора особенно важно при:

  • восстановлении соединения;
  • долговременных подписках;
  • сложной маршрутизации;
  • работе с несколькими брокерами;
  • диагностике.

Пример:

client.subscribe('/topic/orders', onOrderMessage, {
    id: 'orders-stream'
});

Конфликт идентификаторов

Два одинаковых id внутри одного соединения приводят к ошибкам.

Некоторые брокеры:

  • закрывают соединение;
  • отклоняют вторую подписку;
  • заменяют предыдущую подписку;
  • возвращают ERROR frame.

Проблемный пример:

client.subscribe('/topic/a', callbackA, {
    id: 'same-id'
});

client.subscribe('/topic/b', callbackB, {
    id: 'same-id'
});

Параметр ack

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

Параметр ack определяет механизм подтверждения доставки сообщений.

Доступные режимы:

Значение Описание
auto Автоматическое подтверждение
client Ручное подтверждение
client-individual Индивидуальное подтверждение

Режим auto

В режиме auto брокер считает сообщение доставленным сразу после отправки клиенту.

client.subscribe('/queue/jobs', message => {
    console.log(message.body);
}, {
    ack: 'auto'
});

Особенности:

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

Механизм работы auto

Схема обработки:

  1. Брокер отправляет сообщение.
  2. Клиент получает пакет.
  3. Сообщение автоматически считается подтверждённым.
  4. Повторная доставка невозможна.

Если приложение аварийно завершится после получения сообщения, брокер уже удалит его из очереди.

Режим client

Режим client требует ручного подтверждения.

client.subscribe('/queue/orders', message => {

    processOrder(message.body);

    message.ack();

}, {
    ack: 'client'
});

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

message.ack();

Групповое подтверждение

В режиме client подтверждение может распространяться сразу на несколько сообщений.

Например:

client.subscribe('/queue/events', message => {

    console.log(message.body);

    message.ack();

}, {
    ack: 'client'
});

Если до текущего сообщения были неподтверждённые сообщения, брокер может подтвердить их одновременно.

Поведение зависит от реализации брокера.

Преимущества client

Ручное подтверждение позволяет:

  • избежать потери сообщений;
  • реализовать повторную обработку;
  • контролировать транзакции;
  • подтверждать только успешные операции;
  • использовать retry-механизмы.

Недостатки client

Основные проблемы:

  • увеличение нагрузки;
  • рост количества сетевых операций;
  • риск накопления неподтверждённых сообщений;
  • необходимость контроля ack/nack;
  • сложность восстановления состояния.

Режим client-individual

В этом режиме каждое сообщение подтверждается отдельно.

client.subscribe('/queue/payments', message => {

    processPayment(message.body);

    message.ack();

}, {
    ack: 'client-individual'
});

Подтверждение одного сообщения не влияет на остальные.

Отличие от client

Сравнение режимов:

Режим Подтверждение
client Может подтверждать группу сообщений
client-individual Подтверждает только одно сообщение

Когда использовать client-individual

Режим особенно полезен:

  • при параллельной обработке;
  • в системах высокой надёжности;
  • при независимых задачах;
  • в финансовых операциях;
  • при сложной retry-логике.

Использование message.nack()

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

client.subscribe('/queue/tasks', message => {

    try {

        executeTask(message.body);

        message.ack();

    } catch (error) {

        message.nack();
    }

}, {
    ack: 'client-individual'
});

nack() сообщает брокеру о неудачной обработке.

Возможные последствия:

  • повторная доставка;
  • отправка в dead-letter queue;
  • удаление сообщения;
  • перенаправление;
  • увеличение счётчика ошибок.

Durable-подписки

Некоторые брокеры поддерживают долговременные подписки.

Пример:

client.subscribe('/topic/notifications', callback, {
    id: 'notifications-sub',
    durable: 'true'
});

Такая подписка сохраняется на стороне брокера даже после отключения клиента.

Durable-подписки в ActiveMQ

Для ActiveMQ часто используются дополнительные параметры:

client.subscribe('/topic/news', callback, {
    id: 'news-subscription',
    durable: 'true',
    'activemq.subscriptionName': 'news-storage'
});

Брокер начинает сохранять сообщения для офлайн-клиента.

Shared-подписки

Некоторые брокеры поддерживают распределённые подписки между несколькими клиентами.

Пример для RabbitMQ:

client.subscribe('/exchange/events', callback, {
    'x-queue-name': 'shared-workers'
});

Несколько клиентов могут обрабатывать сообщения из одной очереди.

Селекторы сообщений

Брокеры JMS-совместимого типа поддерживают message selectors.

Пример:

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

Подписчик получит только сообщения с нужным условием.

Сложные селекторы

Допустимы логические выражения:

client.subscribe('/topic/orders', callback, {
    selector: "priority = 'high' AND region = 'EU'"
});

Или:

client.subscribe('/topic/orders', callback, {
    selector: "amount > 1000"
});

Browser-specific параметры

Некоторые брокеры используют нестандартные заголовки:

client.subscribe('/topic/logs', callback, {
    browser: 'true'
});

Подобные параметры зависят исключительно от брокера сообщений.

STOMP.js не интерпретирует их содержимое.

Prefetch-настройки

Многие брокеры позволяют ограничивать количество сообщений в буфере клиента.

Пример:

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

Или:

client.subscribe('/queue/jobs', callback, {
    'activemq.prefetchSize': '5'
});

Влияние prefetch

Prefetch определяет:

  • сколько сообщений брокер отправит заранее;
  • размер локального буфера;
  • скорость обработки;
  • объём памяти;
  • уровень параллелизма.

Малый prefetch

Небольшие значения:

{
    'prefetch-count': '1'
}

Преимущества:

  • равномерное распределение;
  • минимизация потерь;
  • снижение нагрузки на память;
  • контроль порядка обработки.

Недостатки:

  • уменьшение throughput;
  • больше сетевых операций.

Большой prefetch

Высокие значения:

{
    'prefetch-count': '1000'
}

Преимущества:

  • высокая производительность;
  • меньше сетевых запросов;
  • эффективный batching.

Недостатки:

  • рост памяти;
  • риск неравномерного распределения;
  • возможная потеря сообщений при сбое клиента.

Подписка с несколькими параметрами

Комплексный пример:

client.subscribe('/queue/orders', message => {

    try {

        processOrder(message.body);

        message.ack();

    } catch (e) {

        message.nack();
    }

}, {
    id: 'orders-worker',
    ack: 'client-individual',
    durable: 'true',
    selector: "priority = 'high'",
    'prefetch-count': '5'
});

Динамическое формирование параметров

Параметры подписки часто собираются программно.

Пример:

const headers = {
    ack: 'client-individual'
};

if (isDurable) {
    headers.durable = 'true';
}

if (selector) {
    headers.selector = selector;
}

client.subscribe(destination, callback, headers);

Переиспользование конфигурации

Распространённая практика — создание шаблонов подписок.

const defaultSubscriptionConfig = {
    ack: 'client-individual',
    durable: 'true'
};

client.subscribe('/queue/a', callbackA, {
    ...defaultSubscriptionConfig,
    id: 'sub-a'
});

client.subscribe('/queue/b', callbackB, {
    ...defaultSubscriptionConfig,
    id: 'sub-b'
});

Проверка параметров

Ошибки в заголовках часто приводят к труднообнаружимым проблемам.

Типичные ошибки:

{
    ack: 'clients'
}

Неверное значение:

{
    ack: 'client'
}

Влияние брокера сообщений

Поддержка параметров зависит от брокера:

Брокер Особенности
RabbitMQ Очереди, prefetch, exchange
ActiveMQ Durable, selectors
Artemis JMS-заголовки
Apollo Специфические subscription headers
HornetQ Расширенные параметры ack

Отладка подписок

STOMP.js позволяет анализировать передаваемые заголовки через debug-функцию.

client.debug = message => {
    console.log(message);
};

Логирование покажет отправляемый SUBSCRIBE frame:

>>> SUBSCRIBE
id:orders-worker
ack:client-individual
destination:/queue/orders
Оглавление