JMS-клиент#

// Создаем фабрику соединений для выполнения клиентского подключения к брокеру, используя URI с параметрами подключения  
ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://0.0.0.0:61616?tcpReceiveBufferSize=1048576;tcpSendBufferSize=1048576");  

// Создаем JMS-соединение с брокером  
Connection connection = cf.createConnection();  

connection.setClientID("jms-core-client-app-id");  

// Регистрируем обработчик ошибок подключения  
connection.setExceptionListener(e -> System.out.println("Ошибка соединения с брокером: " + e.getMessage()));  

// Создаем объект JMS-очередь для очереди `test-address::queue-1` на брокере  
Queue queue = ActiveMQJMSClient.createQueue("test-address::queue-1");  

// Создаем JMS-сессию для поддержки работы с потреблением и поставкой событий брокеру  
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // Поддержка автоматических подтверждений получения сообщений  
  
// Создаем поставщик (продюсер) сообщений в адрес на брокере  
MessageProducer producer = session.createProducer(address);  
  
// Создаем потребителель (консьюмер) сообщений из очереди на брокере  
MessageConsumer messageConsumer = session.createConsumer(queue);  
  
// Создаем текстовое сообщение для отправки в очередь  
TextMessage message = session.createTextMessage("This is a text message");  

// Создаем потребителель (консьюмер) сообщений из очереди на брокере  
MessageConsumer messageConsumer = session.createConsumer(queue);  
  
// Стартуем соединение, для того чтобы консьюмер смог начать получать сообщения с брокера  
connection.start();  
  
// Вычитываем сообщение из брокера с помощью консьюмера  
TextMessage messageReceived = (TextMessage) messageConsumer.receive(5000);  

// Регистрируем обработчик асинхронного получения сообщений с брокера  
consumer.setMessageListener(new MessageListener() {  
	@Override
	public void onMessage(Message message) {
		try {
			var messageReceived = (TextMessage) message;
			System.out.println("Консьюмер получил с брокера сообщение: " + messageReceived.getText());
			messageReceived.acknowledge(); // Подтверждаем факт успешной вычитки сообщения
		} catch (JMSException e) {
			throw new RuntimeException(e); 
		} 
	}
});  

// Создаем объект JMS-очередь для полного имени (FQQN) очереди `test-address::queue-1` на брокере  
Queue queue = ActiveMQJMSClient.createQueue("test-address::queue-1");  
  
// Создаем продюсер сообщений в очередь  
MessageProducer producer = session.createProducer(queue);  
  
// Отправляем сообщение с помощью продюсера на заранее привязанную к продюсеру очередь `test-address::queue-1`  
producer.send(message);  

// Создаем фабрику соединений для выполнения клиентского подключения к брокеру, используя URI с параметрами подключения  
ConnectionFactory cf = new ActiveMQConnectionFactory(URL + ";confirmationWindowSize=1");  
connection = cf.createConnection();  
Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);  
TextMessage message = session.createTextMessage("This is a text message"); // Создаем текстовое сообщение для отправки в очередь  

// Создаем асинхронный обработчик уведомлений о получении сообщений на брокере  
CompletionListener completionListener= new CompletionListener() {  
	@Override
	public void onCompletion(Message message) {
		try {
			logger.info("Для сессии [ClientID={}] получено асинхронное уведомление об успешном приеме на брокере сообщения: ID={}", clientId, message.getJMSMessageID());
		} catch (JMSException e) {
			throw new RuntimeException(e);
		}
	}  
	@Override
	public void onException(Message message, Exception e) {
		try {
			logger.info("Для сессии [ClientID={}] получено асинхронное уведомление о неудачном приеме на брокере сообщения: ID={}", clientId, message.getJMSMessageID());
		} catch (JMSException ex) {
			throw new RuntimeException(ex);
		}
	}
};  

producer.send(message, completionListener);  

// Отправляем сообщение с помощью продюсера с кастомным обработчиком в асинхронном режиме  
producer.send(message, new CompletionListener() {  
	@Override
	public void onCompletion(Message message) {
		logger.info("Получено уведомление об успешном приеме на брокере сообщения: {}", message);
	}
	@Override
	public void onException(Message message, Exception e) {
		logger.info("Получено уведомление о неудачном приеме на брокере сообщения: {}, {}", message, e.getMessage());
	}
 });  

// Создаем текстовое UTF-8 сообщение для отправки в очередь  
TextMessage messageStrUtf8 = session.createTextMessage("Тело текстового сообщения в кодировке UTF-8");  

// Сохраняем текст в кодировке Windows-1251 как массив байт  
byte[] win1251Body = ("Тело бинарного сообщения - текст в кодировке Windows-1251").getBytes("Windows-1251");  
// Создаем бинарное сообщение  
BytesMessage messageStrWin1251 = session.createBytesMessage();  
// Заполняем тело бинарного массивом байт  
messageStrWin1251.writeBytes(win1251Body);  

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";  
// Создаем экземпляр фабрики соединений для выполнения клиентского подключения к брокеру, используя URI с параметрами подключения  
ConnectionFactory cf = new ActiveMQConnectionFactory(url);  

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

import org.apache.activemq.artemis.api.jms.ActiveMQJMSClient;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;

public class ExJms1Run {
    private static final Logger logger = LoggerFactory.getLogger(ExJms1Run.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 = "jms_1";
    public static final String EX_DESCRIPTION = "JMS-клиент с отправкой и вычиткой текстового сообщения";

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

        Connection connection = null;
        try {
            // Создаем фабрику соединений для выполнения клиентского подключения к брокеру, используя URI с параметрами подключения
            ConnectionFactory cf = new ActiveMQConnectionFactory(URL);

            // Создаем JMS-соединение с брокером
            connection = cf.createConnection();

            // Уникальный идентификатор, который поможет в дальнейшем во время отладки и мониторинга клиентского приложения
            connection.setClientID("jms-core-client-app-id");

            // Создаем JMS-сессию для поддержки работы с потреблением и поставкой событий брокеру
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // Поддержка автоматических подтверждений получения сообщений

            // Создаем объект JMS-очередь для полного имени (FQQN) очереди `test-address::queue-1` на брокере
            Queue queue = ActiveMQJMSClient.createQueue(FQQN);

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

            // Создаем потребитель (консьюмер) сообщений из очереди на брокере
            MessageConsumer messageConsumer = session.createConsumer(queue);

            // Создаем текстовое сообщение для отправки в очередь
            TextMessage message = session.createTextMessage("This is a text message");

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

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

            // Вычитываем сообщение из брокера с помощью консьюмера
            TextMessage messageReceived = (TextMessage) messageConsumer.receive(5000);
            logger.info("Консьюмер получил с брокера сообщение: {}", messageReceived.getText());
        } finally {
            if (connection != null) {
                // Закрываем соединение после окончания работы с брокером
                connection.close();
            }
        }
    }

}

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

import org.slf4j.Logger;
import ru.sbt.ss.artemis.clients.jms.ExJms1Run;
import ru.sbt.ss.artemis.demo.DemoExample;

public class ExJms1 implements DemoExample {
    public String getNumber() {
        return ExJms1Run.EX_NUMBER;
    }

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

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

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

import org.apache.activemq.artemis.api.jms.ActiveMQJMSClient;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;

public class ExJms2Run {
    private static final Logger logger = LoggerFactory.getLogger(ExJms2Run.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 = "jms_2";
    public static final String EX_DESCRIPTION = "JMS-клиент с синхронной и асинхронной отправкой, обработчики MessageListener и CompletionListener";

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

        Connection connection = null;
        try {
            var setConfirmationWindowSize = ";confirmationWindowSize=1";  // размер окна подтверждения должен быть >= 0, чтобы заработала пересылки уведомлений в обработчик CompletionListener

            // Создаем фабрику соединений для выполнения клиентского подключения к брокеру, используя URI с параметрами подключения
            ConnectionFactory cf = new ActiveMQConnectionFactory(URL + setConfirmationWindowSize);

            // Создаем JMS-соединение с брокером
            connection = cf.createConnection();

            // Уникальный идентификатор, который поможет в дальнейшем во время отладки и мониторинга клиентского приложения
            connection.setClientID("jms-core-client-app-id");
            var clientId = connection.getClientID();

            // Регистрируем обработчик ошибок подключения
            connection.setExceptionListener(e -> logger.error("Ошибка соединения с брокером: {}", e.getMessage()));

            // Создаем асинхронную JMS-сессию для поддержки работы с потреблением и поставкой событий брокеру
            Session asyncSession = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); // Требуется явное подтверждение факта получения сообщений

            // Создаем объект JMS-очередь для полного имени (FQQN) очереди `test-address::queue-1` на брокере
            Queue queue = ActiveMQJMSClient.createQueue(FQQN);

            // Создаем продюсер сообщений в очередь с асинхронной отправкой сообщений
            MessageProducer asyncSessionProducer = asyncSession.createProducer(queue);

            // Создаем асинхронный обработчик уведомлений о получении сообщений на брокере
            CompletionListener completionListener= new CompletionListener() {
                @Override
                public void onCompletion(Message message) {
                    try {
                        logger.info("Для сессии [ClientID={}] получено асинхронное уведомление об успешном приеме на брокере сообщения: ID={}", clientId, message.getJMSMessageID());
                    } catch (JMSException e) {
                        throw new RuntimeException(e);
                    }
                }

                @Override
                public void onException(Message message, Exception e) {
                    try {
                        logger.info("Для сессии [ClientID={}] получено асинхронное уведомление о неудачном приеме на брокере сообщения: ID={}", clientId, message.getJMSMessageID());
                    } catch (JMSException ex) {
                        throw new RuntimeException(ex);
                    }
                }
            };

            // Создаем консьюмер сообщений из очереди `test-address::queue-1` на брокере
            MessageConsumer messageConsumer = asyncSession.createConsumer(queue);

            // Регистрируем обработчик асинхронного получения сообщений с брокера
            messageConsumer.setMessageListener(message -> {
                try {
                    var messageReceived = (TextMessage) message;
                    logger.info("Консьюмер сессии [ClientID={}] получил асинхронно с брокера сообщение: ID={}, содержимое={}", clientId, message.getJMSMessageID(), messageReceived.getText());
                    messageReceived.acknowledge(); // Подтверждаем факт успешной вычитки сообщения
                } catch (JMSException e) {
                    throw new RuntimeException(e);
                }
            });

            // Создаем текстовое сообщение №1 для отправки в очередь
            TextMessage message1 = asyncSession.createTextMessage("This is a text message 1");
            // Создаем текстовое сообщение №2 для отправки в очередь
            TextMessage message2 = asyncSession.createTextMessage("This is a text message 2");

            logger.info("Продюсер сессии [ClientID={}] готов асинхронно отправить на брокер ПЕРВОЕ сообщение: ID={}, содержимое={}", clientId, message1.getJMSMessageID(), message1.getText());
            // Отправляем сообщение с помощью продюсера с обработчиком уведомлений в асинхронном режиме
            asyncSessionProducer.send(message1, completionListener);
            logger.info("Продюсер сессии [ClientID={}] завершил асинхронную отправку на брокер сообщения: ID={}, содержимое={}", clientId, message1.getJMSMessageID(), message1.getText());

            logger.info("Продюсер сессии [ClientID={}] готов асинхронно отправить на брокер ВТОРОЕ сообщение: ID={}, содержимое={}", clientId, message2.getJMSMessageID(), message2.getText());
            // Отправляем сообщение с помощью продюсера с кастомным обработчиком уведомлений в асинхронном режиме
            asyncSessionProducer.send(message2, new CompletionListener() {
                @Override
                public void onCompletion(Message message) {
                    logger.info("Получено уведомление об успешном приеме на брокере сообщения: {}", message);
                }
                @Override
                public void onException(Message message, Exception e) {
                    logger.info("Получено уведомление о неудачном приеме на брокере сообщения: {}, {}", message, e.getMessage());
                }
            });
            logger.info("Продюсер сессии [ClientID={}] завершил асинхронную отправку на брокер сообщения: ID={}, содержимое={}", clientId, message2.getJMSMessageID(), message2.getText());

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

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

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

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

}

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

import org.slf4j.Logger;
import ru.sbt.ss.artemis.clients.jms.ExJms2Run;
import ru.sbt.ss.artemis.demo.DemoExample;

public class ExJms2 implements DemoExample {
    public String getNumber() {
        return ExJms2Run.EX_NUMBER;
    }

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

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

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

import org.apache.activemq.artemis.api.jms.ActiveMQJMSClient;
import org.apache.activemq.artemis.jms.client.ActiveMQBytesMessage;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.apache.activemq.artemis.jms.client.ActiveMQMessage;
import org.apache.activemq.artemis.jms.client.ActiveMQTextMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;
import java.io.UnsupportedEncodingException;
import java.time.Instant;

public class ExJms3Run {
    private static final Logger logger = LoggerFactory.getLogger(ExJms3Run.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 = "jms_3";
    public static final String EX_DESCRIPTION = "JMS-клиент с примерами работы с сообщениями разных типов (текстовые и бинарные)";

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

        Connection connection = null;
        try {
            var setConfirmationWindowSize = ";confirmationWindowSize=1";  // размер окна подтверждения должен быть >= 0, чтобы заработала пересылки уведомлений в обработчик CompletionListener

            // Создаем фабрику соединений для выполнения клиентского подключения к брокеру, используя URI с параметрами подключения
            ConnectionFactory cf = new ActiveMQConnectionFactory(URL + setConfirmationWindowSize);

            // Создаем JMS-соединение с брокером
            connection = cf.createConnection();

            // Создаем JMS-сессию для поддержки работы с потреблением и поставкой событий брокеру
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // Создаем объект JMS-очередь для полного имени (FQQN) очереди `test-address::queue-1` на брокере
            Queue queue = ActiveMQJMSClient.createQueue(FQQN);

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

            // Создаем консьюмер сообщений из очереди `test-address::queue-1` на брокере
            MessageConsumer messageConsumer = session.createConsumer(queue);

            // Регистрируем обработчик асинхронного получения сообщений с брокера
            messageConsumer.setMessageListener(message -> {
                try {
                    switch (((ActiveMQMessage) message).getType()) {
                        case ActiveMQTextMessage.TYPE: {
                            var textMessage = (TextMessage) message;
                            logger.info("Консьюмер получил с брокера текстовое сообщение: ID={}, содержимое={}", textMessage.getJMSMessageID(), textMessage);
                            break;
                        }
                        case ActiveMQBytesMessage.TYPE: {
                            var bytesMessage = (BytesMessage) message;
                            var bytes = new byte[(int) bytesMessage.getBodyLength()];
                            bytesMessage.readBytes(bytes);
                            try {
                                String strWin1251 = new String(bytes, "Windows-1251");
                                logger.info("Консьюмер получил с брокера бинарное сообщение: ID={}, содержимое={}", bytesMessage.getJMSMessageID(), strWin1251);
                            } catch (UnsupportedEncodingException e) {
                                throw new RuntimeException(e);
                            }
                            break;
                        }
                        default: {
                            logger.info("Консьюмер получил с брокера {} сообщение: ID={}", message.getJMSType(), message.getJMSMessageID());
                        }
                    }
                } catch (JMSException e) {
                    throw new RuntimeException(e);
                }
            });

            // Создаем текстовое UTF-8 сообщение для отправки в очередь
            String utf8Body = "Тело текстового сообщения в кодировке UTF-8 создано " + Instant.now();
            TextMessage messageStrUtf8 = session.createTextMessage(utf8Body);

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

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

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

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

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

            // Небольшая пауза перед завершением работы
            logger.info("Небольшая пауза перед завершением работы");
            Thread.sleep(10000);
        } finally {
            if (connection != null) {
                // Закрываем соединение после окончания работы с брокером
                connection.stop();
                connection.close();
            }
        }
    }

}

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

import org.slf4j.Logger;
import ru.sbt.ss.artemis.clients.jms.ExJms3Run;
import ru.sbt.ss.artemis.demo.DemoExample;

public class ExJms3 implements DemoExample {
    public String getNumber() {
        return ExJms3Run.EX_NUMBER;
    }

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

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

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

import org.apache.activemq.artemis.api.core.client.*;
import org.apache.activemq.artemis.api.jms.ActiveMQJMSClient;
import org.apache.activemq.artemis.api.jms.ActiveMQJMSConstants;
import org.apache.activemq.artemis.jms.client.ActiveMQBytesMessage;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.apache.activemq.artemis.jms.client.ActiveMQMessage;
import org.apache.activemq.artemis.jms.client.ActiveMQTextMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;
import java.io.*;
import java.nio.charset.StandardCharsets;
import java.time.Instant;

public class ExJms4Run {
    private static final Logger logger = LoggerFactory.getLogger(ExJms4Run.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 = "jms_4";
    public static final String EX_DESCRIPTION = "JMS-клиент с примерами работы с большими сообщениями";

    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

        Connection connection = null;
        try {
            var setConfirmationWindowSize = ";confirmationWindowSize=1";                 // Размер окна подтверждения должен быть >= 0, чтобы заработала пересылки уведомлений в обработчик CompletionListener
            var setCompressionLevel = ";compressionLevel=" + COMPRESSION_LEVEL;          // Устанавливаем уровень сжатия сообщений
            var setCompressLargeMessage = ";compressLargeMessage=" + COMPRESS_LARGE_MSG; // Включаем сжатие больших сообщений
            var setCustomMinLargeMessageSize = ";minLargeMessageSize=" + LARGE_MSG_SIZE; // Устанавливаем кастомный минимальный размер большого сообщения в байтах (по умолчанию 100KiB)

            // Создаем фабрику соединений для выполнения клиентского подключения к брокеру, используя URI с параметрами подключения
            ConnectionFactory cf = new ActiveMQConnectionFactory(URL
                    + setConfirmationWindowSize
                    + setCompressionLevel
                    + setCompressLargeMessage
                    + setCustomMinLargeMessageSize
            );

            // Создаем JMS-соединение с брокером
            connection = cf.createConnection();

            // Создаем JMS-сессию для поддержки работы с потреблением и поставкой событий брокеру
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // Создаем объект JMS-очередь для полного имени (FQQN) очереди `test-address::queue-1` на брокере
            Queue queue = ActiveMQJMSClient.createQueue(FQQN);

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

            // Создаем консьюмер сообщений из очереди `test-address::queue-1` на брокере
            MessageConsumer messageConsumer = session.createConsumer(queue);

            // Регистрируем обработчик асинхронного получения сообщений с брокера
            messageConsumer.setMessageListener(message -> {
                try {
                    switch (((ActiveMQMessage) message).getType()) {
                        case ActiveMQTextMessage.TYPE: {
                            var textMessage = (TextMessage) message;
                            logger.info("Консьюмер получил с брокера текстовое сообщение: id={}, сообщение={}, \nтело={}", textMessage.getJMSMessageID(), textMessage, textMessage.getBody(String.class));
                            break;
                        }
                        case ActiveMQBytesMessage.TYPE: {
                            var bytesMessage = (BytesMessage) message;
                            var bytes = new byte[(int) bytesMessage.getBodyLength()];
                            bytesMessage.readBytes(bytes);
                            try {
                                String strWin1251 = new String(bytes, "Windows-1251");
                                logger.info("Консьюмер получил с брокера бинарное сообщение: id={}, сообщение={}, \nтело={}", bytesMessage.getJMSMessageID(), bytesMessage, strWin1251);
                            } catch (UnsupportedEncodingException e) {
                                throw new RuntimeException(e);
                            }
                            break;
                        }
                        default: {
                            logger.info("Консьюмер получил с брокера сообщение: ID={}, MSG={}", message.getJMSMessageID(), message);
                        }
                    }
                } catch (JMSException e) {
                    throw new RuntimeException(e);
                }
            });

            // Создаем БОЛЬШОЕ текстовое UTF-8 сообщение, размером больше, чем задано в настройке minLargeMessageSize
            // Необходимо учитывать, что при включении сжатия (compressLargeMessage=true), фактический размер сообщения станет меньше!!!
            // Также помним, что каждый символ UTF-8 занимает 2 байта в очереди на брокере!!!
            String utf8Text = "Тело текстового большого сообщения в кодировке UTF-8 ";
            String utf8LargeBody = utf8Text.repeat((int) (COMPRESS_FACTOR * LARGE_MSG_SIZE / utf8Text.length() / 2)); // доводим размер сообщения до БОЛЬШОГО (> minLargeMessageSize)
            TextMessage messageStrUtf8 = session.createTextMessage(utf8LargeBody);
            var utf8MessageSize = messageStrUtf8.getBody(String.class).length();
            logger.info("Создано большое текстовое сообщение, тело (до сжатия) размером: {} байт (размер в очереди - {})", utf8MessageSize, utf8MessageSize * 2);

            // Создаем БОЛЬШОЕ бинарное сообщение с текстом в кодировке Windows-1251 для отправки в очередь в виде массива байт
            // Необходимо учитывать, что при включении сжатия (compressLargeMessage=true), фактический размер сообщения станет меньше!!!
            String win1251Text = "Тело бинарного большого сообщения с текстом в кодировке Windows-1251 ";
            String win1251LargeBodyText = win1251Text.repeat((int) (COMPRESS_FACTOR * LARGE_MSG_SIZE / win1251Text.length())); // доводим размер сообщения до БОЛЬШОГО (> minLargeMessageSize)
            byte[] win1251LargeBodyBytes = win1251LargeBodyText.getBytes("Windows-1251");
            BytesMessage messageStrWin1251 = session.createBytesMessage();
            messageStrWin1251.writeBytes(win1251LargeBodyBytes);
            logger.info("Создано большое бинарное сообщение, тело (до сжатия) размером: {} байт", win1251LargeBodyBytes.length);

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

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

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

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

            logger.info("Небольшая пауза ({}сек) перед завершением работы", delaySec);
            Thread.sleep(delaySec * 1000);
        } finally {
            if (connection != null) {
                // Закрываем соединение после окончания работы с брокером
                connection.stop();
                connection.close();
            }
        }
    }

}

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

import org.slf4j.Logger;
import ru.sbt.ss.artemis.clients.jms.ExJms4Run;
import ru.sbt.ss.artemis.demo.DemoExample;

public class ExJms4 implements DemoExample {
    public String getNumber() {
        return ExJms4Run.EX_NUMBER;
    }

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

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

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

import org.apache.activemq.artemis.api.jms.ActiveMQJMSClient;
import org.apache.activemq.artemis.api.jms.ActiveMQJMSConstants;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;

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

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

    // Предполагается, что на брокере предварительно должен быть создан адрес `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 = "jms_5";
    public static final String EX_DESCRIPTION = "JMS-клиент с разными режимами коммитов вычитки сообщений";

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

        Connection connection = null;
        try {
            // Создаем фабрику соединений для выполнения клиентского подключения к брокеру, используя URI с параметрами подключения
            ConnectionFactory cf = new ActiveMQConnectionFactory(URL);

            // Создаем JMS-соединение с брокером
            connection = cf.createConnection();

            // Уникальный идентификатор, который поможет в дальнейшем во время отладки и мониторинга клиентского приложения
            connection.setClientID("jms-core-client-app-id");

            // Создаем JMS-сессию для поддержки работы с потреблением и поставкой событий брокеру
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
            // Session session = connection.createSession(false, Session.DUPS_OK_ACKNOWLEDGE);
            // Session session = connection.createSession(false, ActiveMQJMSConstants.INDIVIDUAL_ACKNOWLEDGE);
            // Session session = connection.createSession(false, ActiveMQJMSConstants.PRE_ACKNOWLEDGE);


            // Создаем объект JMS-очередь для полного имени (FQQN) очереди `test-address::queue-1` на брокере
            Queue queue = ActiveMQJMSClient.createQueue(FQQN);

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

            // Создаем потребитель (консьюмер) сообщений из очереди на брокере
            MessageConsumer messageConsumer = session.createConsumer(queue);

            // Создаем текстовое сообщение для отправки в очередь
            TextMessage message = session.createTextMessage("This is a text message");

            // Отправляем сообщения с помощью продюсера
            for (var i = 1; i <= 10; i++) {
                producer.send(message);
                logger.info("Продюсер отправил на брокер сообщение №{}: {}", i, message.getText());
            }

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

            // Вычитываем сообщения из брокера с помощью консьюмера
            var groupSize = 5;
            for (var i = 1; i <= 10; i++) {
                message = (TextMessage) messageConsumer.receive();
                logger.info("Консьюмер получил с брокера сообщение №{}: {}", i, message.getText());
                if (i % groupSize == 0) {
                    logger.info("Готовимся вызвать acknowledge() для сообщения №{}: {}", i, message.getText());
                    message.acknowledge();
                    logger.info("Завершен вызов acknowledge() для сообщения №{}: {}", i, message.getText());
                }
            }
        } finally {
            if (connection != null) {
                // Закрываем соединение после окончания работы с брокером
                connection.close();
            }
        }
    }

}

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

import org.slf4j.Logger;
import ru.sbt.ss.artemis.clients.jms.ExJms5Run;
import ru.sbt.ss.artemis.demo.DemoExample;

public class ExJms5 implements DemoExample {
    public String getNumber() {
        return ExJms5Run.EX_NUMBER;
    }

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

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