CORE-клиенты#

ServerLocator locator = ActiveMQClient.createServerLocator("tcp://0.0.0.0:61616");

ClientSessionFactory factory =  locator.createSessionFactory();

// Создаем сессию для поддержки работы с получением и отправкой событий брокеру
ClientSession session = factory.createSession();

// Создаем поставщик сообщений (продюсер) в адрес `test-address` на брокере
ClientProducer producer = session.createProducer("test-address");

// Создаем потребителель сообщений (консьюмер) из очереди `queue-1` на брокере 
ClientConsumer consumer = session.createConsumer("queue-1");

// Создаем объект-сообщение для отправки в очередь
ClientMessage message = session.createMessage(true);

ClientSession session = factory.createSession(null, null, false, true, true, locator.isPreAcknowledge(), locator.getAckBatchSize(), "core-client-app-id");

// Устанавливаем размер окна подтверждения в положительное значение, чтобы заработала пересылка уведомлений в обработчик SendAcknowledgementHandler
locator.setConfirmationWindowSize(1000);

// Создаем сессию
ClientSession session = locator.createSessionFactory().createSession();

// Регистрируем обработчик уведомлений о получении сообщений на брокере
session.setSendAcknowledgementHandler(message ->
        System.out.println("Получено уведомление об успешном приеме на брокере сообщения: " + message)
);

// Создаем сессию для поддержки работы с потреблением и поставкой событий брокеру
ClientSession session = factory.createSession();

// Создаем поставщик сообщений (продюсер) в адрес `test-address` на брокере
ClientProducer producer = session.createProducer("test-address");

// Отправляем сообщение с помощью продюсера на заранее привязанный к продюсеру адрес `test-address`
producer.send(message);

// Создаем поставщик сообщений (продюсер)
ClientProducer producer = session.createProducer();

// Отправляем сообщение с помощью продюсера с указанием адреса назначения `test-address`
producer.send("test-address", message);

// Задает максимальное количество отправленных сообщений, созданных на заданной фабрике
ServerLocator.setProducerMaxRate(int)
// Задает размер окна для ограничения потока продюсеров, создаваемых на заданной фабрике
ServerLocator.setProducerWindowSize(int)

// Включаем блокировку на отправку сообщений брокеру
locator.setBlockOnDurableSend(true);
locator.setBlockOnNonDurableSend(true);

ClientSessionFactory factory =  locator.createSessionFactory();
ClientSession session = factory.createSession();

// Создаем синхронный продюсер
ClientProducer syncProducer = session.createProducer("test-address");

// Отправляем сообщение с помощью продюсера в синхронном режиме
syncProducer.send(message);

// Отправляем сообщение с помощью продюсера с кастомным обработчиком в асинхронном режиме
syncSessionProducer.send(message, recievedMessage ->
        System.out.println("Получено уведомление об успешном приеме на брокере сообщения: " + recievedMessage)

// Создаем потребителель сообщений (консьюмер) из очереди `queue-1` на брокере
ClientConsumer consumer = session.createConsumer("queue-1");

// Стартуем сессию, для того чтобы консьюмер смог начать получать сообщения с брокера
session.start();

// Вычитываем сообщение из брокера с помощью консьюмера в синхронном режиме
ClientMessage msgReceived = consumer.receive();

// Регистрируем обработчик асинхронного получения сообщений с брокера
consumer.setMessageHandler(message ->
       System.out.println("Консьюмер получил с брокера сообщение: " + message.getBodyBuffer().readString())
);

// Создаем текстовое UTF-8 сообщение для отправки в очередь
ClientMessage messageStrUtf8 = session.createMessage(ClientMessage.TEXT_TYPE, true);
String utf8Body = "Тело текстового сообщения в кодировке UTF-8";
TextMessageUtil.writeBodyText( ((CoreMessage) messageStrUtf8).getBodyBuffer(), new SimpleString(utf8Body));

// Создаем бинарное сообщение с текстом в кодировке Windows-1251 для отправки в очередь в виде массива байт
ClientMessage messageStrWin1251 = session.createMessage(ClientMessage.BYTES_TYPE, true);
byte[] win1251Body = "Тело бинарного сообщения - текст в кодировке Windows-1251".getBytes("Windows-1251");
messageStrWin1251.getBodyBuffer().writeBytes(win1251Body);

// Формируем тело большого сообщения через потоки
byte[] msgBody = "Тело большого сообщения ".repeat(2500).getBytes(StandardCharsets.UTF_8);
ClientMessage largeMessage = session.createMessage(true);
InputStream inputStream = new ByteArrayInputStream(msgBody);
largeMessage.setBodyInputStream(inputStream);

// Отправляем сообщение на брокер
producer.send(largeMessage);

ClientMessage messageReceived = consumer.receive();

// Получаем тело большого сообщения через потоки
OutputStream outputStream = new ByteArrayOutputStream();
messageReceived.saveToOutputStream(outputStream);

String url = "tcp://0.0.0.0:61616?protocols=CORE;supportAdvisory=false;suppressInternalManagementObjects=false;tcpReceiveBufferSize=1048576;tcpSendBufferSize=1048576;useEpoll=true;sslEnabled=false;verifyHost=false;needClientAuth=true";
// Создаем экземпляр локатора с помощью URl
ServerLocator locator = ActiveMQClient.createServerLocator(url);

String url = "tcp://0.0.0.0:61616?protocols=CORE;supportAdvisory=false;suppressInternalManagementObjects=false;tcpReceiveBufferSize=1048576;tcpSendBufferSize=1048576;useEpoll=true;sslEnabled=false;verifyHost=false;needClientAuth=true";
// Создаем экземпляр локатора с помощью URl
ServerLocator locator = ActiveMQClient.createServerLocator(url);
// Настраиваем дополнительные параметры подключения
locator.setConfirmationWindowSize(1000); // размер окна подтверждения
locator.setMaxRetryInterval(30000);      // максимальный интервал между попытками переподключения
locator.setConsumerMaxRate(100);         // максимальная скорость потребления сообщений
locator.setProducerMaxRate(200);         // максимальная скорость производства сообщений
// Создаем фабрику сессий для подключения к брокеру
ClientSessionFactory factory =  locator.createSessionFactory();
// После создания инициализации фабрики вызов сеттеров локатора уже не допустим!!!
locator.setGroupID("group-id");          // будет выброшено исключение IllegalStateException: Cannot set attribute on SessionFactory after it has been used 

ClientSession session = factory.createSession(
        null,
        null,
        false,
        false, // autoCommitSends
        false, // autoCommitAcks
        false, // preAcknowledge
        0,     // ackBatchSize
        "session-client-id"
);

ClientSession transactSession = factory.createTransactedSession();

package ru.sbt.ss.artemis.clients.core;

import org.apache.activemq.artemis.api.core.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ExCore1Run {
    private static final Logger logger = LoggerFactory.getLogger(ExCore1Run.class);

    // Предполагается, что брокер Artemis запущен на локальной машине на открытом порту (plain TCP)
    public final static String URL = "tcp://0.0.0.0:61616";

    // Предполагается, что на брокере предварительно должен быть создан адрес `test-address` с очередью `queue-1`
    public static String ADDRESS = "test-address";
    public static String QUEUE = "queue-1";
    public static String FQQN = ADDRESS + "::" + QUEUE;

    public static final String EX_NUMBER = "core_1";
    public static final String EX_DESCRIPTION = "CORE-клиент с отправкой и вычиткой текстового сообщения";

    public static void main(String[] args) throws Exception {
        logger.info("{} ({})", EX_DESCRIPTION, EX_NUMBER);

        // Создаем экземпляр локатора для обнаружения брокера
        ServerLocator locator = ActiveMQClient.createServerLocator(URL);
        locator.setPreAcknowledge(true); // Поддержка автоматических подтверждений получения сообщений

        // Создаем фабрику сессий для выполнения клиентского подключения к брокеру
        ClientSessionFactory factory =  locator.createSessionFactory();

        // Создаем сессию для поддержки работы с потреблением и поставкой событий брокеру
        ClientSession session = factory.createSession();

        // Создаем поставщик сообщений (продюсер) в адрес `test-address` на брокере
        ClientProducer producer = session.createProducer(ADDRESS);

        // Создаем потребитель сообщений (консьюмер) из очереди `queue-1` на брокере
        ClientConsumer consumer = session.createConsumer(QUEUE);

        // Создаем текстовое сообщение для отправки в очередь
        ClientMessage message = session.createMessage(true);
        message.getBodyBuffer().writeString("Тело тестового сообщения");

        // Отправляем сообщение с помощью продюсера
        producer.send(message);
        logger.info("Продюсер отправил на брокер сообщение: {}", message.getBodyBuffer().readString());

        // Стартуем сессию, для того чтобы консьюмер смог начать получать сообщения с брокера
        session.start();

        // Вычитываем сообщение из брокера с помощью консьюмера
        ClientMessage msgReceived = consumer.receive();
        logger.info("Консьюмер получил с брокера сообщение: {}", msgReceived.getBodyBuffer().readString());

        // Закрываем сессию после окончания работы с брокером
        session.close();
    }

}

package ru.sbt.ss.artemis.clients.core.demo;

import org.slf4j.Logger;
import ru.sbt.ss.artemis.demo.DemoExample;
import ru.sbt.ss.artemis.clients.core.ExCore1Run;

public class ExCore1 implements DemoExample {
    public String getNumber() {
        return ExCore1Run.EX_NUMBER;
    }

    @Override
    public String getDescription() {
        return ExCore1Run.EX_DESCRIPTION;
    }

    @Override
    public void run(Logger logger) throws Exception {
        ExCore1Run.main(null);
    }
}

package ru.sbt.ss.artemis.clients.core;

import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.client.*;
import org.apache.activemq.artemis.core.message.impl.CoreMessage;
import org.apache.activemq.artemis.reader.TextMessageUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Instant;

public class ExCore2Run {
    private static final Logger logger = LoggerFactory.getLogger(ExCore2Run.class);

    // Предполагается, что брокер Artemis запущен на локальной машине на открытом порту (plain TCP)
    public final static String URL = "tcp://0.0.0.0:61616";

    // Предполагается, что на брокере предварительно должен быть создан адрес `test-address` с очередью `queue-1`
    public static String ADDRESS = "test-address";
    public static String QUEUE = "queue-1";
    public static String FQQN = ADDRESS + "::" + QUEUE;

    public static final String EX_NUMBER = "core_2";
    public static final String EX_DESCRIPTION = "CORE-клиент с синхронной и асинхронной отправкой, обработчики MessageHandler и SendAcknowledgementHandler";

    public static void main(String[] args) throws Exception {
        logger.info("{} ({})", EX_DESCRIPTION, EX_NUMBER);

        // Создаем экземпляр локатора для обнаружения брокера
        ServerLocator locator = ActiveMQClient.createServerLocator(URL);

        // Настраиваем дополнительные параметры подключения к брокеру
        locator.setConfirmationWindowSize(1);    // размер окна подтверждения должен быть >= 0, чтобы заработала пересылки уведомлений в обработчик SendAcknowledgementHandler
        locator.setMaxRetryInterval(30000);      // максимальный интервал между попытками переподключения
        locator.setConsumerMaxRate(100);         // максимальная скорость потребления сообщений
        locator.setProducerMaxRate(200);         // максимальная скорость производства сообщений
        locator.setBlockOnDurableSend(true);     // включена блокировка на отправку персистентных сообщений
        locator.setBlockOnNonDurableSend(true);  // включена блокировка на отправку неперсистентных сообщений

        // Создаем фабрику сессий для выполнения клиентского подключения к брокеру
        ClientSessionFactory factory =  locator.createSessionFactory();

        // Создаем сессию для поддержки работы с потреблением и поставкой событий брокеру в асинхронном режиме
        var asyncSessionId = "async-app"; // ClientId для сессии с асинхронной отправкой и вычиткой сообщений
        ClientSession asyncSession = factory.createSession(
                null,
                null,
                false,
                true,
                true,
                locator.isPreAcknowledge(),
                locator.getAckBatchSize(),
                asyncSessionId
        );

        // Создаем сессию для поддержки работы с поставкой событий брокеру в синхронном режиме
        var syncSessionId = "sync-app"; // ClientId для сессии с синхронной отправкой
        ClientSession syncSession = factory.createSession(
                null,
                null,
                false,
                true,
                true,
                locator.isPreAcknowledge(),
                locator.getAckBatchSize(),
                syncSessionId
        );

        // Регистрируем асинхронный обработчик уведомлений о получении сообщений на брокере
        asyncSession.setSendAcknowledgementHandler(new SendAcknowledgementHandler() {
            @Override
            public void sendAcknowledged(Message message) {
                logger.info("Для сессии [ClientID={}] получено асинхронное уведомление об успешном приеме на брокере сообщения: ID={}", asyncSessionId, message.getMessageID());
            }

            @Override
            public void sendFailed(Message message, Exception e) {
                logger.info("Для сессии [ClientID={}] получено асинхронное уведомление о неудачном приеме на брокере сообщения: ID={}", asyncSessionId, message.getMessageID());
                SendAcknowledgementHandler.super.sendFailed(message, e);
            }
        });

        // Создаем продюсер для асинхронной сессии
        ClientProducer asyncSessionProducer = asyncSession.createProducer();

        // Создаем продюсер для синхронной сессии
        ClientProducer syncSessionProducer = syncSession.createProducer(ADDRESS);

        // Создаем консьюмер для сессии с асинхронной отправкой сообщений
        ClientConsumer asyncSessionConsumer = asyncSession.createConsumer(QUEUE);

        // Регистрируем обработчик асинхронного получения сообщений с брокера
        asyncSessionConsumer.setMessageHandler(new MessageHandler() {
            @Override
            // Каждый раз, когда консьюмер получает сообщение, будет вызываться метод `onMessage()` обработчика.
            public void onMessage(ClientMessage message) {
                try {
                    logger.info("Консьюмер сессии [ClientID={}] получил асинхронно с брокера сообщение: ID={}, содержимое={}", asyncSessionId, message.getMessageID(), TextMessageUtil.readBodyText(message.getDataBuffer()));
                    message.acknowledge(); // Подтверждаем факт успешной вычитки сообщения
                } catch (ActiveMQException e) {
                    throw new RuntimeException(e);
                }
            }

            @Override
            public void onMessageExpired(ClientMessage message) {
                logger.info("Консьюмер сессии [ClientID={}] получил асинхронно с брокера просроченное сообщение: ID={}, содержимое={}", asyncSessionId, message.getMessageID(), TextMessageUtil.readBodyText(message.getDataBuffer()));
                MessageHandler.super.onMessageExpired(message);
            }
        });

        // Создаем текстовое сообщение для отправки в очередь
        ClientMessage message = asyncSession.createMessage(ClientMessage.TEXT_TYPE, true);
        TextMessageUtil.writeBodyText( ((CoreMessage) message).getBodyBuffer(), new SimpleString("Тело тестового сообщения создано " + Instant.now()));

        logger.info("Продюсер сессии [ClientID={}] готов синхронно отправить на брокер сообщение: ID={}, содержимое={}",  syncSessionId, message.getMessageID(), TextMessageUtil.readBodyText(message.getDataBuffer()));

        Thread.sleep(5000);

        // Отправляем сообщение с помощью продюсера в синхронном режиме
        syncSessionProducer.send(message);
        logger.info("Для сессии [ClientID={}] получено синхронное уведомление об успешном приеме на брокере сообщения: ID={}, содержимое={}", syncSessionId, message.getMessageID(), TextMessageUtil.readBodyText(message.getDataBuffer()));

        logger.info("Продюсер сессии [ClientID={}] готов асинхронно отправить на брокер сообщение: ID={}, содержимое={}", syncSessionId, message.getMessageID(), TextMessageUtil.readBodyText(message.getDataBuffer()));
        // Отправляем сообщение с помощью продюсера с кастомным обработчиком в асинхронном режиме
        syncSessionProducer.send(message, recievedMessage -> logger.info("Для сессии [ClientId={}] получено уведомление об успешном приеме на брокере сообщения: {}", syncSessionId, recievedMessage));
        logger.info("Продюсер сессии [ClientID={}] завершил асинхронную отправку на брокер сообщения: ID={}, содержимое={}", syncSessionId, message.getMessageID(), TextMessageUtil.readBodyText(message.getDataBuffer()));

        logger.info("Продюсер сессии [ClientID={}] готов асинхронно отправить на брокер сообщение: ID={}, содержимое={}", asyncSessionId, message.getMessageID(), TextMessageUtil.readBodyText(message.getDataBuffer()));
        // Отправляем сообщение с помощью продюсера в синхронном режиме
        asyncSessionProducer.send(ADDRESS, message);
        logger.info("Продюсер сессии [ClientID={}] завершил асинхронную отправку на брокер сообщения: ID={}, содержимое={}", asyncSessionId, message.getMessageID(), TextMessageUtil.readBodyText(message.getDataBuffer()));

        // Небольшая пауза перед стартом соединения для начала вычитки сообщений из очереди
        logger.info("Небольшая пауза перед стартом сессии (подключением консьюмера)");
        Thread.sleep(1000);

        // Стартуем асинхронную сессию, для того чтобы консьюмер смог начать получать сообщения с брокера
        asyncSession.start();
        logger.info("Стартовала асинхронная сессия!");

        // Стартовать синхронную сессию не обязательно, так как с ней не связан ни один консьюмер
        // syncSession.start();

        logger.info("А теперь можно поискать в консоли управления брокера (на вкладках Connections, Consumers, Producers) информацию о клиенте по ClientID={} и ClientID={} ...", syncSessionId, asyncSessionId);
        // Задержка на время для поиска информации о клиенте в консоли управления брокера
        Thread.sleep(30000);

        logger.info("Все, 30 секунд истекло, наш клиент закрыл сессии и их ClientID исчезли из консоли управления брокера :(");
        // Закрываем сессии после окончания работы с брокером
        asyncSession.stop();
        asyncSession.close();

        syncSession.close();
    }

}

package ru.sbt.ss.artemis.clients.core.demo;

import org.slf4j.Logger;
import ru.sbt.ss.artemis.demo.DemoExample;
import ru.sbt.ss.artemis.clients.core.ExCore2Run;

public class ExCore2 implements DemoExample {
    public String getNumber() {
        return ExCore2Run.EX_NUMBER;
    }

    @Override
    public String getDescription() {
        return ExCore2Run.EX_DESCRIPTION;
    }

    @Override
    public void run(Logger logger) throws Exception {
        ExCore2Run.main(null);
    }
}

package ru.sbt.ss.artemis.clients.core;

import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.client.*;
import org.apache.activemq.artemis.core.message.impl.CoreMessage;
import org.apache.activemq.artemis.reader.TextMessageUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;
import java.time.Instant;

public class ExCore3Run {
    private static final Logger logger = LoggerFactory.getLogger(ExCore3Run.class);

    // Предполагается, что брокер Artemis запущен на локальной машине на открытом порту (plain TCP)
    public final static String URL = "tcp://0.0.0.0:61616";

    // Предполагается, что на брокере предварительно должен быть создан адрес `test-address` с очередью `queue-1`
    public static String ADDRESS = "test-address";
    public static String QUEUE = "queue-1";
    public static String FQQN = ADDRESS + "::" + QUEUE;

    public static final String EX_NUMBER = "core_3";
    public static final String EX_DESCRIPTION = "CORE-клиент с примерами работы с сообщениями разных типов (текстовые и бинарные)";

    public static void main(String[] args) throws Exception {
        logger.info("{} ({})", EX_DESCRIPTION, EX_NUMBER);

        // Создаем экземпляр локатора для обнаружения брокера
        ServerLocator locator = ActiveMQClient.createServerLocator(URL);
        locator.setPreAcknowledge(true); // Поддержка автоматических подтверждений получения сообщений

        // Создаем фабрику сессий для выполнения клиентского подключения к брокеру
        ClientSessionFactory factory =  locator.createSessionFactory();

        // Создаем сессию для поддержки работы с потреблением и поставкой событий брокеру
        ClientSession session = factory.createSession();

        // Создаем поставщик сообщений (продюсер) в адрес `test-address` на брокере
        ClientProducer producer = session.createProducer(ADDRESS);

        // Создаем потребителель сообщений (консьюмер) из очереди `queue-1` на брокере
        ClientConsumer consumer = session.createConsumer(QUEUE);

        // Регистрируем обработчик асинхронного получения сообщений с брокера
        consumer.setMessageHandler(message -> {
            switch (message.getType()) {
                case ClientMessage.TEXT_TYPE: {
                    logger.info("Консьюмер получил с брокера текстовое сообщение: ID={}, содержимое={}", message.getMessageID(), TextMessageUtil.readBodyText(message.getDataBuffer()));
                    break;
                }
                case ClientMessage.BYTES_TYPE: {
                    var bytes = new byte[message.getBodyBuffer().readableBytes()];
                    message.getBodyBuffer().readBytes(bytes);
                    try {
                        String strWin1251 = new String(bytes, "Windows-1251");
                        logger.info("Консьюмер получил с брокера бинарное сообщение: ID={}, содержимое={}", message.getMessageID(), strWin1251);
                    } catch (UnsupportedEncodingException e) {
                        throw new RuntimeException(e);
                    }
                    break;
                }
                default: {
                    logger.info("Консьюмер получил с брокера сообщение: ID={}", message.getMessageID());
                }
            }
        });

        // Создаем текстовое UTF-8 сообщение для отправки в очередь
        ClientMessage messageStrUtf8 = session.createMessage(ClientMessage.TEXT_TYPE, true);
        String utf8Body = "Тело текстового сообщения в кодировке UTF-8 создано " + Instant.now();
        TextMessageUtil.writeBodyText( ((CoreMessage) messageStrUtf8).getBodyBuffer(), new SimpleString(utf8Body));

        // Создаем бинарное сообщение с текстом в кодировке Windows-1251 для отправки в очередь в виде массива байт
        ClientMessage messageStrWin1251 = session.createMessage(ClientMessage.BYTES_TYPE, true);
        byte[] win1251Body = ("Тело бинарного сообщения - текст в кодировке Windows-1251 создано " + Instant.now()).getBytes("Windows-1251");
        messageStrWin1251.getBodyBuffer().writeBytes(win1251Body);

        // Отправляем сообщение с помощью продюсера
        producer.send(messageStrUtf8);
        logger.info("Продюсер отправил на брокер сообщение: {}", TextMessageUtil.readBodyText(messageStrUtf8.getDataBuffer()));

        // Отправляем сообщение с помощью продюсера
        producer.send(messageStrWin1251);
        logger.info("Продюсер отправил на брокер сообщение: {}", messageStrWin1251);

        // Стартуем сессию, для того чтобы консьюмер смог начать получать сообщения с брокера
        session.start();

        // Закрываем сессию после окончания работы с брокером
        session.close();
    }

}

package ru.sbt.ss.artemis.clients.core.demo;

import org.slf4j.Logger;
import ru.sbt.ss.artemis.demo.DemoExample;
import ru.sbt.ss.artemis.clients.core.ExCore3Run;

public class ExCore3  implements DemoExample {
    public String getNumber() {
        return ExCore3Run.EX_NUMBER;
    }

    @Override
    public String getDescription() {
        return ExCore3Run.EX_DESCRIPTION;
    }

    @Override
    public void run(Logger logger) throws Exception {
        ExCore3Run.main(null);
    }
}

package ru.sbt.ss.artemis.clients.core;

import org.apache.activemq.artemis.api.core.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.nio.charset.StandardCharsets;

public class ExCore4Run {
    private static final Logger logger = LoggerFactory.getLogger(ExCore4Run.class);

    // Предполагается, что брокер Artemis запущен на локальной машине на открытом порту (plain TCP)
    public final static String URL = "tcp://0.0.0.0:61616";

    // Предполагается, что на брокере предварительно должен быть создан адрес `test-address` с очередью `queue-1`
    public static String ADDRESS = "test-address";
    public static String QUEUE = "queue-1";
    public static String FQQN = ADDRESS + "::" + QUEUE;

    public static final String EX_NUMBER = "core_4";
    public static final String EX_DESCRIPTION = "CORE-клиент с примерами работы с большими сообщениями через Streaming Core API";

    public static void main(String[] args) throws Exception {
        logger.info("{} ({})", EX_DESCRIPTION, EX_NUMBER);

        int COMPRESSION_LEVEL = 0;         // Степень сжатия сообщений в пределах от `0` до `9`, по умолчанию используется значение `-1` (это соответствует уровню 6-7).
        int LARGE_MSG_SIZE = 10000;        // Кастомный минимальный размер большого сообщения в байтах
        boolean COMPRESS_LARGE_MSG = true; // Включать сжатие больших сообщений?
        double COMPRESS_FACTOR = !(COMPRESS_LARGE_MSG && COMPRESSION_LEVEL != 0)? 1.1 : 300; // Эмпирический коэффициент увеличения размера исходного сообщения, чтобы после компрессии с уровнем `5` размер был > minLargeMessageSize

        // Создаем экземпляр локатора для обнаружения брокера
        ServerLocator locator = ActiveMQClient.createServerLocator(URL);
        locator.setPreAcknowledge(true);                       // Поддержка автоматических подтверждений получения сообщений
        locator.setCompressionLevel(COMPRESSION_LEVEL);        // Уровень сжатия сообщений
        locator.setCompressLargeMessage(COMPRESS_LARGE_MSG);   // Включаем сжатие больших сообщений
        locator.setMinLargeMessageSize(LARGE_MSG_SIZE);        // Кастомный минимальный размер большого сообщения

        // Создаем фабрику сессий для выполнения клиентского подключения к брокеру
        ClientSessionFactory factory =  locator.createSessionFactory();

        // Создаем сессию для поддержки работы с потреблением и поставкой событий брокеру
        ClientSession session = factory.createSession();

        // Создаем поставщик сообщений (продюсер) в адрес `test-address` на брокере
        ClientProducer producer = session.createProducer(ADDRESS);

        // Создаем потребителель сообщений (консьюмер) из очереди `queue-1` на брокере
        ClientConsumer consumer = session.createConsumer(QUEUE);

        // Формируем тело первого большого сообщения через потоки (Streaming Core API)
        String utf8Text = "Тело большого сообщения № 1 с текстом в кодировке UTF-8  ";
        String utf8TextLargeBodyText = utf8Text.repeat((int) (COMPRESS_FACTOR * LARGE_MSG_SIZE / utf8Text.length())); // доводим размер сообщения до БОЛЬШОГО (> minLargeMessageSize)
        byte[] msgBody1 = utf8TextLargeBodyText.getBytes(StandardCharsets.UTF_8);
        ClientMessage largeMessageWithStream = session.createMessage(true);
        ByteArrayInputStream inputStream = new ByteArrayInputStream(msgBody1);
        largeMessageWithStream.setBodyInputStream(inputStream);
        logger.info("Создано большое сообщение №1 с текстом в кодировке UTF-8, тело (поток) размером: {} байт", largeMessageWithStream.getBodySize());

        // Формируем тело второго большого сообщения через буфер
        String win1251Text = "Тело большого сообщения №2 с текстом в кодировке Windows-1251 ";
        String win1251LargeBodyText = win1251Text.repeat((int) (COMPRESS_FACTOR * LARGE_MSG_SIZE / win1251Text.length())); // доводим размер сообщения до БОЛЬШОГО (> minLargeMessageSize)
        byte[] win1251LargeBodyBytes = win1251LargeBodyText.getBytes("Windows-1251");
        ClientMessage largeMessageWithBuffer = session.createMessage(true);
        largeMessageWithBuffer.getBodyBuffer().writeBytes(win1251LargeBodyBytes);
        logger.info("Создано большое сообщение №2 с текстом в кодировке Windows-1251, тело (буфер) размером: {} байт", largeMessageWithBuffer.getBodySize());

        // Отправляем оба сообщения на брокер
        producer.send(largeMessageWithStream);
        logger.info("Продюсер отправил на брокер большое сообщение №1: {}", largeMessageWithStream);
        producer.send(largeMessageWithBuffer);
        logger.info("Продюсер отправил на брокер большое сообщение №2: {}", largeMessageWithBuffer);

        // Небольшая пауза перед стартом соединения для начала вычитки сообщений из очереди
        // Можно убедиться, что сообщения успешно попали в очередь и имеют признак большого сообщения ("largeMessage": true)
        var delaySec = 10;
        logger.info("Небольшая пауза ({}сек) перед стартом сессии (подключением консьюмера)", delaySec);
        Thread.sleep(delaySec * 1000);

        // Стартуем сессию, для того чтобы консьюмер смог начать получать сообщения с брокера
        session.start();

        // Вычитываем сообщение №1 из брокера с помощью консьюмера
        ClientMessage messageReceived1 = consumer.receive();

        // Получаем тело большого сообщения через потоки
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        messageReceived1.saveToOutputStream(outputStream);
        var text1 = outputStream.toString(StandardCharsets.UTF_8);
        logger.info("Консьюмер получил с брокера большое сообщение с текстом в кодировке UTF-8({}байт): {}", text1.length(), text1);

        // Вычитываем сообщение №2 из брокера с помощью консьюмера
        ClientMessage messageReceived2 = consumer.receive();

        // Получаем тело большого сообщения через буфер
        var bytes = new byte[messageReceived2.getBodyBuffer().readableBytes()];
        messageReceived2.getBodyBuffer().readBytes(bytes);
        var text2 = new String(bytes, "Windows-1251");
        logger.info("Консьюмер получил с брокера большое сообщение с текстом в кодировке Windows-1251({}байт): {}", text2.length(), text2);

        logger.info("Небольшая пауза ({}сек) перед завершением работы", delaySec);
        Thread.sleep(delaySec * 1000);

        // Закрываем сессию после окончания работы с брокером
        session.close();
    }

}

package ru.sbt.ss.artemis.clients.core.demo;

import org.slf4j.Logger;
import ru.sbt.ss.artemis.demo.DemoExample;
import ru.sbt.ss.artemis.clients.core.ExCore4Run;

public class ExCore4 implements DemoExample {
    public String getNumber() {
        return ExCore4Run.EX_NUMBER;
    }

    @Override
    public String getDescription() {
        return ExCore4Run.EX_DESCRIPTION;
    }

    @Override
    public void run(Logger logger) throws Exception {
        ExCore4Run.main(null);
    }
}

package ru.sbt.ss.artemis.clients.core;

import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.client.*;
import org.apache.activemq.artemis.core.message.impl.CoreMessage;
import org.apache.activemq.artemis.reader.TextMessageUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Instant;

public class ExCore5Run {
    private static final Logger logger = LoggerFactory.getLogger(ExCore5Run.class);

    // Предполагается, что брокер Artemis запущен на локальной машине на открытом порту (plain TCP)
    public final static String URL = "tcp://0.0.0.0:61616";

    // Предполагается, что на брокере предварительно должен быть создан адрес `test-address` с очередью `queue-1`
    public static String ADDRESS = "test-address";
    public static String QUEUE = "queue-1";
    public static String FQQN = ADDRESS + "::" + QUEUE;

    public static final String EX_NUMBER = "core_5";
    public static final String EX_DESCRIPTION = "CORE-клиент с транзакциями";

    public static void main(String[] args) throws Exception {
        logger.info("{} ({})", EX_DESCRIPTION, EX_NUMBER);

        // Создаем экземпляр локатора для обнаружения брокера
        ServerLocator locator = ActiveMQClient.createServerLocator(URL);

        // Настраиваем дополнительные параметры подключения к брокеру
        locator.setAckBatchSize(1); // Устанавливает размер буфера для подтверждений

        // Создаем фабрику сессий для выполнения клиентского подключения к брокеру
        ClientSessionFactory factory =  locator.createSessionFactory();

        // Создаем транзакционную сессию для поддержки работы с потреблением и поставкой событий брокеру
        // ClientSession implicitTransactSession = factory.createTransactedSession(); // сессия с транзакцией
        var transactSessionId = "transact-app"; // ClientId для сессии с транзакцией

        ClientSession transactSession = factory.createSession(
                null,
                null,
                false,
                false, // autoCommitSends
                false, // autoCommitAcks
                locator.isPreAcknowledge(), // preAcknowledge
                locator.getAckBatchSize(),  //ackBatchSize
                transactSessionId
        );

        // Создаем продюсер для транзакционной сессии
        ClientProducer producer = transactSession.createProducer(ADDRESS);

        // Создаем консьюмер для транзакционной сессии
        ClientConsumer consumer = transactSession.createConsumer(QUEUE);

        // Создаем текстовые сообщения для отправки в очередь
        ClientMessage message1 = transactSession.createMessage(ClientMessage.TEXT_TYPE, true);
        TextMessageUtil.writeBodyText( ((CoreMessage) message1).getBodyBuffer(), new SimpleString("Тело тестового сообщения №1 создано " + Instant.now()));
        ClientMessage message2 = transactSession.createMessage(ClientMessage.TEXT_TYPE, true);
        TextMessageUtil.writeBodyText( ((CoreMessage) message2).getBodyBuffer(), new SimpleString("Тело тестового сообщения №2 создано " + Instant.now()));
        ClientMessage message3 = transactSession.createMessage(ClientMessage.TEXT_TYPE, true);
        TextMessageUtil.writeBodyText( ((CoreMessage) message3).getBodyBuffer(), new SimpleString("Тело тестового сообщения №3 создано " + Instant.now()));

        // Стартуем транзакционную сессию
        transactSession.start();

        // Отправляем сообщения
        logger.info("Продюсер сессии [ClientID={}] готов асинхронно отправить на брокер сообщение: ID={}, содержимое={}", transactSessionId, message1.getMessageID(), TextMessageUtil.readBodyText(message1.getDataBuffer()));
        // Отправляем сообщение №1 с помощью продюсера в асинхронном режиме
        producer.send(message1);
        logger.info("Продюсер сессии [ClientID={}] завершил асинхронную отправку на брокер сообщения: ID={}, содержимое={}", transactSessionId, message1.getMessageID(), TextMessageUtil.readBodyText(message1.getDataBuffer()));

        logger.info("Продюсер сессии [ClientID={}] готов асинхронно отправить на брокер сообщение: ID={}, содержимое={}", transactSessionId, message2.getMessageID(), TextMessageUtil.readBodyText(message2.getDataBuffer()));
        // Отправляем сообщение №2 с помощью продюсера в асинхронном режиме
        producer.send(message2);
        logger.info("Продюсер сессии [ClientID={}] завершил асинхронную отправку на брокер сообщения: ID={}, содержимое={}", transactSessionId, message2.getMessageID(), TextMessageUtil.readBodyText(message2.getDataBuffer()));

        logger.info("Продюсер сессии [ClientID={}] готов асинхронно отправить на брокер сообщение: ID={}, содержимое={}", transactSessionId, message3.getMessageID(), TextMessageUtil.readBodyText(message3.getDataBuffer()));
        // Отправляем сообщение №3 с помощью продюсера в асинхронном режиме
        producer.send(message3);
        logger.info("Продюсер сессии [ClientID={}] завершил асинхронную отправку на брокер сообщения: ID={}, содержимое={}", transactSessionId, message3.getMessageID(), TextMessageUtil.readBodyText(message3.getDataBuffer()));

        // Коммитим транзакцию (отправку сообщений)
        logger.info("Коммитим транзакцию (отправку сообщений)");
        transactSession.commit();
        logger.info("Закоммичена транзакция (отправка сообщений)");

        // Вычитываем сообщение №1
        logger.info("Консьюмер начал вычитку");
        ClientMessage msgReceived = consumer.receive();
        logger.info("Консьюмер получил с брокера сообщение: ID={}, содержимое={}", msgReceived.getMessageID(), TextMessageUtil.readBodyText(msgReceived.getDataBuffer()));
        // Подтверждаем вычитку сообщения №1
        msgReceived.acknowledge();
        logger.info("Подтверждено (ack) получение сообщения: ID={}, содержимое={}", msgReceived.getMessageID(), TextMessageUtil.readBodyText(msgReceived.getDataBuffer()));

        // Коммитим транзакцию (получение сообщения №1 было подтверждено)
        logger.info("Коммитим транзакцию (получение сообщения №1)");
        transactSession.commit();
        logger.info("Закоммичена транзакция (получение сообщения №1)");

        // Вычитываем сообщение №2
        logger.info("Консьюмер начал вычитку");
        msgReceived = consumer.receive();
        logger.info("Консьюмер получил с брокера сообщение: ID={}, содержимое={}", msgReceived.getMessageID(), TextMessageUtil.readBodyText(msgReceived.getDataBuffer()));
        // Подтверждаем вычитку сообщения №2
        msgReceived.acknowledge();
        logger.info("Подтверждено (ack) получение сообщения: ID={}, содержимое={}", msgReceived.getMessageID(), TextMessageUtil.readBodyText(msgReceived.getDataBuffer()));

        // Откатываем сессию, что приведет к переотправке сообщений с брокера
        logger.info("Откатываем транзакцию");
        transactSession.rollback();
        logger.info("Завершено откатывание транзакции");

        // Вычитываем сообщение №2 заново
        logger.info("Консьюмер начал вычитку");
        msgReceived = consumer.receive();
        logger.info("Консьюмер получил заново (redelivering) после отката транзакции сообщение: ID={}, содержимое={}", msgReceived.getMessageID(), TextMessageUtil.readBodyText(msgReceived.getDataBuffer()));
        msgReceived.acknowledge();
        logger.info("Подтверждено (ack) получение сообщения: ID={}, содержимое={}", msgReceived.getMessageID(), TextMessageUtil.readBodyText(msgReceived.getDataBuffer()));
        // Вычитываем сообщение №3
        logger.info("Консьюмер начал вычитку");
        msgReceived = consumer.receive();
        logger.info("Консьюмер получил после отката транзакции сообщение: ID={}, содержимое={}", msgReceived.getMessageID(), TextMessageUtil.readBodyText(msgReceived.getDataBuffer()));
        msgReceived.acknowledge();
        logger.info("Подтверждено (ack) получение сообщения: ID={}, содержимое={}", msgReceived.getMessageID(), TextMessageUtil.readBodyText(msgReceived.getDataBuffer()));

        // Коммитим транзакцию (получение сообщений №2 и №3 было подтверждено)
        logger.info("Коммитим транзакцию (повторное получение сообщений №2 и №3)");
        transactSession.commit();
        logger.info("Закоммичена транзакция (повторное получение сообщений №2 и №3)");

        logger.info("А теперь можно поискать в консоли управления брокера (на вкладках Connections, Consumers, Producers) информацию о клиенте по ClientID={} ...", transactSessionId);
        // Задержка на время для поиска информации о клиенте в консоли управления брокера
        Thread.sleep(30000);

        logger.info("Все, 30 секунд истекло, наш клиент закрыл транзакционную сессию и ее ClientID исчез из консоли управления брокера :(");
        // Закрываем сессии после окончания работы с брокером
        transactSession.stop();
        transactSession.close();
    }

}

package ru.sbt.ss.artemis.clients.core.demo;

import org.slf4j.Logger;
import ru.sbt.ss.artemis.demo.DemoExample;
import ru.sbt.ss.artemis.clients.core.ExCore5Run;

public class ExCore5 implements DemoExample {
    public String getNumber() {
        return ExCore5Run.EX_NUMBER;
    }

    @Override
    public String getDescription() {
        return ExCore5Run.EX_DESCRIPTION;
    }

    @Override
    public void run(Logger logger) throws Exception {
        ExCore5Run.main(null);
    }
}