Начало транзакции

Транзакция в протоколе STOMP позволяет объединить несколько операций отправки и подтверждения сообщений в единый атомарный блок. Пока транзакция не завершена командой COMMIT, брокер рассматривает все действия как незавершённые. При отмене через ABORT изменения откатываются.

В STOMP.js транзакции используются для:

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

Транзакционный режим особенно важен при работе с финансовыми операциями, заказами, синхронизацией данных, системами логирования и обработкой событий.

Метод begin()

Начало транзакции в STOMP.js выполняется методом begin() объекта клиента.

Базовый синтаксис:

const transaction = client.begin();

После вызова метода создаётся объект транзакции, содержащий уникальный идентификатор и методы управления.

Пример:

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

const client = new Client({
    brokerURL: 'ws://localhost:15674/ws'
});

client.onConn ect = () => {

    const transaction = client.begin();

    console.log(transaction.id);
};

client.activate();

Что происходит после начала транзакции

После вызова begin() клиент отправляет брокеру STOMP-фрейм:

BEGIN
transaction:tx-1

^@

Где:

  • BEGIN — команда протокола;
  • transaction — идентификатор транзакции;
  • tx-1 — уникальный ID.

Все последующие операции могут быть связаны с этим идентификатором.

Объект транзакции

Метод begin() возвращает специальный объект транзакции.

Типичная структура:

const tx = client.begin();

Объект содержит:

  • id
  • commit()
  • abort()

Пример:

const tx = client.begin();

console.log(tx.id);

tx.commit();

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

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

Пример автоматически созданного ID:

tx-0
tx-1
tx-2

Это предотвращает конфликты между несколькими активными транзакциями.

Указание собственного идентификатора

STOMP.js позволяет задавать ID транзакции вручную.

Пример:

const tx = client.begin('order-processing');

Тогда будет отправлен фрейм:

BEGIN
transaction:order-processing

^@

Пользовательские идентификаторы полезны:

  • при логировании;
  • при трассировке операций;
  • при интеграции с backend-системами;
  • при диагностике ошибок.

Несколько одновременных транзакций

STOMP допускает параллельное существование нескольких транзакций.

Пример:

const tx1 = client.begin('payments');
const tx2 = client.begin('notifications');

После этого операции можно разделять:

client.publish({
    destination: '/queue/payments',
    body: 'Payment completed',
    headers: {
        transaction: tx1.id
    }
});

client.publish({
    destination: '/queue/notifications',
    body: 'User notified',
    headers: {
        transaction: tx2.id
    }
});

Каждая транзакция обрабатывается независимо.

Транзакция и отправка сообщений

Сообщение становится частью транзакции только при указании заголовка transaction.

Пример:

const tx = client.begin();

client.publish({
    destination: '/queue/tasks',
    body: 'Task 1',
    headers: {
        transaction: tx.id
    }
});

Без заголовка сообщение будет отправлено немедленно вне транзакции.

Отправка нескольких сообщений в одной транзакции

Одна транзакция может включать множество сообщений.

Пример:

const tx = client.begin();

client.publish({
    destination: '/queue/logs',
    body: 'Log entry 1',
    headers: {
        transaction: tx.id
    }
});

client.publish({
    destination: '/queue/logs',
    body: 'Log entry 2',
    headers: {
        transaction: tx.id
    }
});

client.publish({
    destination: '/queue/logs',
    body: 'Log entry 3',
    headers: {
        transaction: tx.id
    }
});

tx.commit();

Брокер применит изменения только после COMMIT.

Использование транзакций с ACK

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

Пример:

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

    const tx = client.begin();

    try {

        processOrder(message);

        message.ack({
            transaction: tx.id
        });

        tx.commit();

    } catch (error) {

        tx.abort();
    }

}, {
    ack: 'client'
});

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

Транзакция и режим ACK

Наиболее часто транзакции используются вместе с:

  • client
  • client-individual

Пример:

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

Режим auto практически не сочетается с транзакционной обработкой, поскольку сообщения подтверждаются автоматически.

Когда транзакция действительно необходима

Транзакции оправданы в следующих сценариях:

Финансовые операции

const tx = client.begin();

sendDebit(tx);
sendCredit(tx);

tx.commit();

Согласованная запись логов

const tx = client.begin();

writeMainLog(tx);
writeAuditLog(tx);

tx.commit();

Групповая доставка событий

const tx = client.begin();

events.forEach(event => {

    client.publish({
        destination: '/topic/events',
        body: JSON.stringify(event),
        headers: {
            transaction: tx.id
        }
    });

});

tx.commit();

Ошибки при начале транзакции

Начало транзакции до подключения

Ошибка:

const tx = client.begin();

при отсутствии соединения.

Правильный вариант:

client.onConn ect = () => {

    const tx = client.begin();
};

Потеря ссылки на транзакцию

Ошибка:

client.begin();

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

Транзакция создаётся, но её ID нигде не используется.

Правильно:

const tx = client.begin();

client.publish({
    destination: '/queue/test',
    body: 'data',
    headers: {
        transaction: tx.id
    }
});

Использование несуществующего transaction ID

Ошибка:

client.publish({
    destination: '/queue/test',
    body: 'data',
    headers: {
        transaction: 'unknown-tx'
    }
});

Брокер может отклонить такую операцию.

Проверка активной транзакции

STOMP.js не хранит глобальный список транзакций. Управление их состоянием полностью лежит на приложении.

Обычно применяется собственная система хранения:

const activeTransactions = new Map();

const tx = client.begin();

activeTransactions.set(tx.id, tx);

Логирование транзакций

При отладке полезно фиксировать:

  • момент создания;
  • transaction ID;
  • сообщения внутри транзакции;
  • время завершения;
  • ошибки.

Пример:

const tx = client.begin();

console.log('Transaction started:', tx.id);

Таймауты транзакций

Сам STOMP-протокол не содержит встроенного механизма таймаута транзакций. Если транзакция не завершена:

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

Поэтому часто реализуют пользовательский таймер:

const tx = client.begin();

const timeout = setTimeout(() => {

    tx.abort();

}, 5000);

Оборачивание транзакций в async/await

Транзакции удобно интегрируются в асинхронный код.

Пример:

async function saveBatch(items) {

    const tx = client.begin();

    try {

        for (const item of items) {

            client.publish({
                destination: '/queue/items',
                body: JSON.stringify(item),
                headers: {
                    transaction: tx.id
                }
            });
        }

        tx.commit();

    } catch (error) {

        tx.abort();

        throw error;
    }
}

Поведение брокеров

Разные брокеры могут по-разному реализовывать транзакции:

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

Наиболее популярные брокеры с поддержкой STOMP:

  • RabbitMQ
  • ActiveMQ
  • Apache Artemis

Внутренний механизм STOMP.js

При вызове:

client.begin();

STOMP.js:

  1. создаёт transaction ID;
  2. формирует BEGIN-фрейм;
  3. отправляет его через WebSocket;
  4. возвращает объект управления транзакцией.

Метод commit() позже отправляет:

COMMIT
transaction:tx-1

^@

А abort():

ABORT
transaction:tx-1

^@

Практический пример полноценной транзакции

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

const client = new Client({
    brokerURL: 'ws://localhost:15674/ws'
});

client.onConn ect = () => {

    const tx = client.begin('batch-import');

    try {

        client.publish({
            destination: '/queue/import',
            body: 'Record 1',
            headers: {
                transaction: tx.id
            }
        });

        client.publish({
            destination: '/queue/import',
            body: 'Record 2',
            headers: {
                transaction: tx.id
            }
        });

        client.publish({
            destination: '/queue/import',
            body: 'Record 3',
            headers: {
                transaction: tx.id
            }
        });

        tx.commit();

        console.log('Transaction committed');

    } catch (error) {

        tx.abort();

        console.error('Transaction aborted', error);
    }
};

client.activate();
Оглавление