Руководство разработчика JMS 2.0 служебной шины Azure

В этом руководстве содержатся подробные сведения, помогающие успешно взаимодействовать с Служебная шина Azure с помощью API службы сообщений Java (JMS) 2.0.

Если вы не знакомы с служебной шиной Azure, ознакомьтесь со следующими статьями.

модель программирования службы сообщений Java (JMS)

Модель программирования API службы сообщений Java описана в следующих разделах:

JMS — стандартные блоки

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

Фабрика подключений

Клиент использует объект фабрики соединений для подключения к поставщику JMS. Фабрика подключений инкапсулирует набор параметров конфигурации подключения, которые определяет администратор.

Каждая фабрика соединений является экземпляром интерфейса ConnectionFactory, интерфейса QueueConnectionFactory или интерфейса TopicConnectionFactory.

Чтобы упростить подключение к Служебная шина Azure, эти интерфейсы реализуются с помощью ServiceBusJmsConnectionFactory, ServiceBusJmsQueueConnectionFactory или ServiceBusJmsTopicConnectionFactory соответственно.

TokenCredential tokenCredential = new DefaultAzureCredentialBuilder().build();

String host = "<YourNamespaceName>.servicebus.windows.net";
ConnectionFactory factory = new ServiceBusJmsConnectionFactory(tokenCredential, host, null); 

TokenCredential tokenCredential = new DefaultAzureCredentialBuilder()
                                                .managedIdentityClientId("<clientIDOfUserAssignedIdentity>")
                                                .build();

String host = "<YourNamespaceName>.servicebus.windows.net";
ConnectionFactory factory = new ServiceBusJmsConnectionFactory(tokenCredential, host, null); 

TokenCredential tokenCredential = new ClientSecretCredentialBuilder()
                .tenantId("")
                .clientId("")
                .clientSecret("")
                .build();

ConnectionFactory factory = new ServiceBusJmsConnectionFactory(SERVICE_BUS_CONNECTION_STRING, null);

Назначение JMS

Назначение — это объект, который клиент использует для указания целевого объекта сообщений, которые он создает, и источника потребляемых сообщений.

Назначения соответствуют сущностям в Служебная шина Azure — очереди (в сценариях 'точка-точка') и топики (в сценариях pub-sub).

Связи

Соединение инкапсулирует виртуальное соединение с поставщиком услуг JMS. Служебная шина Azure представляет подключение с отслеживанием состояния между приложением и служебной шиной Azure через AMQP.

Создайте подключение из фабрики подключений, как показано в следующем примере:

Connection connection = factory.createConnection();

Сеансы

Сеанс — это однопоточный контекст для создания и потребления сообщений. Используйте его для создания сообщений, производителей сообщений и потребителей. Он также предоставляет транзакционный контекст для объединения операций отправки и получения в атомарную единицу работы.

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

Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);

Режимы сеанса

Создайте сеанс с любым из следующих режимов.

Если режим сеанса не указан, значение по умолчанию Session.AUTO_ACKNOWLEDGE.

JMSContext

JMSContext объединяет функциональные возможности, предоставляемые объектом подключения и сеанса. Вы создаете его из объекта фабрики соединений.

JMSContext context = connectionFactory.createContext();

Режимы JMSContext

Как и объект Session , можно создать JMSContext с теми же режимами подтверждения, что и в режимах сеанса.

JMSContext context = connectionFactory.createContext(JMSContext.AUTO_ACKNOWLEDGE);

Если режим не указан, по умолчанию используется JMSContext.AUTO_ACKNOWLEDGE.

Производители сообщений JMS

Производитель сообщений — это объект, создаваемый с помощью JMSContext или сеанса. Используйте его для отправки сообщений в место назначения.

Его можно создать как автономный объект, как показано в следующем примере:

JMSProducer producer = context.createProducer();

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

context.createProducer().send(destination, message);

Потребители сообщений JMS

Потребитель сообщения — это объект, создаваемый JMSContext или сеансом. Используйте его для получения сообщений, отправленных в место назначения. Создайте его, как показано в следующем примере:

JMSConsumer consumer = context.createConsumer(dest);

Синхронные приемы через метод receive()

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

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

Message m = consumer.receive();
Message m = consumer.receive(0);

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

Message m = consumer.receive(1000); // time out after one second.

Асинхронное получение сообщений с прослушивателями JMS

Прослушиватель сообщений — это объект, используемый для асинхронной обработки сообщений на пункте назначения. Он реализует MessageListener интерфейс, содержащий onMessage метод, в котором должна жить конкретная бизнес-логика.

Необходимо создать экземпляр объекта прослушивателя сообщений и зарегистрировать его для определенного потребителя сообщения с помощью setMessageListener метода.

Listener myListener = new Listener();
consumer.setMessageListener(myListener);

Потребление из тем

Вы создаете потребители сообщений JMS для пункта назначения, которое может быть очередью или темой.

Потребители в очередях — это просто клиентские объекты, которые живут в контексте сеанса (и подключения) между клиентским приложением и Служебная шина Azure.

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

• Клиентский объект, который существует в контексте сеанса (или JMSContext) и
• Подписка, которая является сущностью в Сервисной шине Azure.

Подписки описаны здесь и могут быть одним из следующих типов:

• общие устойчивые подписки;
• общие неустойчивые подписки;
• Неразделяемые устойчивые подписки
• Необщие неустойчивые подписки

Браузеры очередей JMS

API JMS предоставляет QueueBrowser объект, который приложение может использовать для просмотра сообщений в очереди и отображения значений заголовков для каждого сообщения.

Вы можете создать браузер очередей с помощью JMSContext, как показано в следующем примере:

QueueBrowser browser = context.createBrowser(queue);

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

Приложения-получатели могут использовать селекторы сообщений для фильтрации получаемых сообщений. Используя селекторы сообщений, принимающее приложение выгружает работу фильтрации сообщений поставщику JMS (в данном случае Служебная шина Azure) вместо того, чтобы взять на себя эту ответственность.

При создании любого из следующих потребителей можно использовать селекторы:

• Общая устойчивая подписка
• Несоотделенные устойчивые подписки
• Общая непостоянная подписка
• Неразделяемая кратковременная подписка
• Потребитель очереди
• Браузер очередей

Запланированные сообщения (задержка доставки)

JMS 2.0 поддерживает планирование сообщения для будущей доставки с помощью метода setDeliveryDelay на MessageProducer или JMSProducer. При установке этого свойства служебная шина принимает сообщение, но только делает его видимым для потребителей после истечения периода задержки.

MessageProducer producer = session.createProducer(queue);

// Schedule a message for delivery 30 seconds from now
producer.setDeliveryDelay(30000);
producer.send(session.createTextMessage("Scheduled message"));

Полный рабочий пример см. в разделе QueueScheduledSend.java в репозитории azure-servicebus-jms-samples.

Выбор и отказоустойчивость фабрики подключений

При использовании ServiceBusJmsConnectionFactory в Spring Boot или других платформах, которые управляют подключениями JMS, выберите правильную оболочку фабрики соединений для отправителей и прослушивателей , чтобы обеспечить надежную операцию.

Рекомендуемая конфигурация

Чего следует избегать слушателям

CachingConnectionFactory В контейнерах прослушивателей также могут возникнуть проблемы, так как кэшированные сессии могут ссылаться на устаревшее базовое подключение. Для прослушивателей сырая фабрика гарантирует, что каждый контейнер может создавать новое подключение независимо.

Параметры по умолчанию Spring Cloud Azure

Если вы используете spring-cloud-azure-starter-servicebus-jms (версия 6.2.0+), стартовый механизм применяет это разделение фабрики по умолчанию.

В более ранних версиях (до 6.2.0) и отправители, и прослушиватели по умолчанию используют ServiceBusJmsConnectionFactory, что приводит к созданию нового подключения для каждой отправки.

Добавление прослушивателя исключений

Без прослушивателя исключений разрыв соединения происходит незаметно. Добавьте jakarta.jms.ExceptionListener и в фабрику отправителя, и в фабрику слушателя для обеспечения наблюдаемости.

connection.setExceptionListener(exception -> {
    log.error("JMS connection error: {}", exception.getMessage(), exception);
});

В Spring Boot задайте прослушиватель исключений ( CachingConnectionFactory для отправителей) и DefaultJmsListenerContainerFactory (для прослушивателей).

Полный рабочий пример, показывающий все эти шаблоны, см. в примере Spring Boot JMS Resilience в репозитории azure-servicebus-jms-samples.

Очереди недоставленных сообщений

Каждая очередь и подписка на тему в Служебная шина Azure имеет связанную очередь отложенных сообщений dead letter queue (DLQ). Система автоматически перемещает в DLQ сообщения, которые она не может доставить или обработать. Например, система перемещает сообщение в DLQ, когда оно превышает максимальное количество доставок или истекает срок его жизни (TTL).

В JMS вы обращаетесь к DLQ в качестве отдельного назначения, создав полный путь и создав JmsQueue его. Специальный API не требуется.

Формат пути DLQ очереди:

<queue-name>/$deadletterqueue

Формат пути DLQ для подписки на топик:

<topic-name>/Subscriptions/<subscription-name>/$deadletterqueue

Пример: использование из очереди недоставленных писем:

import org.apache.qpid.jms.JmsQueue;

// Construct the DLQ path for a queue named "orders"
String dlqPath = "orders/$deadletterqueue";
JmsQueue dlqDestination = new JmsQueue(dlqPath);

// Create a consumer on the DLQ and receive messages
MessageConsumer dlqConsumer = session.createConsumer(dlqDestination);
Message message = dlqConsumer.receive(5000);

Сообщения с недоставленными буквами включают свойства метаданных, описывающие, почему сообщение было недоставлено:

Прочитайте эти свойства с помощью message.getStringProperty():

String reason = message.getStringProperty("DeadLetterReason");
String description = message.getStringProperty("DeadLetterErrorDescription");

Полный рабочий пример см. в разделе QueueDeadLetterReceive.java в репозитории azure-servicebus-jms-samples.

Сопоставление управления AMQP и операций шины служб

Вот как распоряжение AMQP преобразуется в операцию служебная шина:

ACCEPTED = 1; -> Complete()
REJECTED = 2; -> DeadLetter()
RELEASED = 3; (just unlock the message in service bus, will then get redelivered)
MODIFIED_FAILED = 4; -> Abandon() which increases delivery count
MODIFIED_FAILED_UNDELIVERABLE = 5; -> Defer()

Сводка

В этом руководстве разработчика показано, как Java клиентские приложения, использующие службу сообщений Java (JMS), могут подключаться к Служебная шина Azure.

Дальнейшие действия

Дополнительные сведения об Служебная шина Azure и сущностях Java Message Service (JMS) см. в следующих источниках:

• Выберите между JMS и собственным пакетом SDK для Служебная шина Azure
• Шина сервисов — очереди, топики и подписки
• Служебная шина — сущности службы сообщений Java
• Поддержка AMQP 1.0 в служебной шине Azure
• Руководство разработчика служебной шины AMQP 1.0
• Начало работы с очередями служебная шина
• API службы сообщений Java (внешний документ Oracle)
• Узнайте, как выполнить миграцию с ActiveMQ на служебную шину