Используйте Java Message Service 1.1 с Служебная шина Azure уровня Standard и AMQP 1.0

Warning

В этой статье рассматривается ограниченная поддержка API Java Message Service (JMS) 1.1 на уровне Standard службы Служебная шина Azure.

Для полной поддержки JMS 2.0 используйте Служебная шина Azure Premium уровня.

Используйте Apache Qpid JMS для отправки и получения сообщений через AMQP 1.0. В примерах используется Qpid напрямую; они не зависят от выведенной из эксплуатации библиотеки com.microsoft.azure:azure-servicebus или внутренних классов клиентской библиотеки Azure.

Начните работу с шиной сообщений

Создайте пространство имён служебная шина и очередь. Пример очереди использует очередь с именем BasicQueue.

Создайте политику совместного доступа с правами отправки и прослушивания. Настройте следующие переменные среды.

Variable Value
SB_SAMPLES_NAMESPACE Хост пространства имён, например example.servicebus.windows.net.
SB_SAMPLES_SAS_KEY_NAME Имя политики общего доступа.
SB_SAMPLES_SAS_KEY Ключ политики совместного доступа.

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

Скачайте клиентскую библиотеку AMQP 1.0 JMS

Установите JDK 11 или более позднюю версию и Maven, затем клонируйте репозиторий примеров служебная шина. Пример очереди JMS включает её зависимости в pom.xml.

В примере используется javax.jms API с клиентом Qpid JMS 1.x. Для информации о версии Qpid и API см. документацию Apache Qpid JMS.

Программируйте приложения на Java

Интерфейс именования и каталогов Java

Java Naming and Directory Interface (JNDI) отделяет логические имена от физических адресов. Провайдер Qpid JNDI определяет фабрику соединений и очередь из свойств конфигурации.

Настройте контекст JNDI и настройте объект ConnectionFactory

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

import javax.jms.ConnectionFactory;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.naming.Context;
import javax.naming.InitialContext;
import java.util.Hashtable;

String namespaceHost = System.getenv("SB_SAMPLES_NAMESPACE");
String sasKeyName = System.getenv("SB_SAMPLES_SAS_KEY_NAME");
String sasKey = System.getenv("SB_SAMPLES_SAS_KEY");
if (namespaceHost == null || namespaceHost.isEmpty()
        || sasKeyName == null || sasKeyName.isEmpty()
        || sasKey == null || sasKey.isEmpty()) {
    throw new IllegalArgumentException("Set the namespace host, policy name, and policy key.");
}

Hashtable<String, String> properties = new Hashtable<>();
properties.put(Context.INITIAL_CONTEXT_FACTORY,
        "org.apache.qpid.jms.jndi.JmsInitialContextFactory");
properties.put("connectionfactory.SBCF",
        "amqps://" + namespaceHost + "?amqp.idleTimeout=120000");
properties.put("queue.QUEUE", "BasicQueue");

Context context = new InitialContext(properties);
ConnectionFactory factory = (ConnectionFactory) context.lookup("SBCF");
Destination queue = (Destination) context.lookup("QUEUE");

Настройка очередей назначения производителя и потребителя

Создайте JMS-соединение, используя имя политики и ключ. Используйте очередь, разрешённую через JNDI, в качестве назначения, вместо того чтобы приводить строку имени очереди к типу Destination.

Connection connection = factory.createConnection(sasKeyName, sasKey);
Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
MessageProducer producer = session.createProducer(queue);

Создайте отдельную сессию для потребителя. Начните его соединение до получения сообщений.

Простое приложение JMS, использующее очередь служебная шина

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

Запуск приложения

Из каталога samples/Java/qpid-jms-client/JmsQueueQuickstart соберите пример. Тест интеграции также использует настроенное пространство имён и очередь.

mvn clean package

Запустите упакованное приложение:

java -jar target/jmsqueuequickstart-1.0.0-jar-with-dependencies.jar

Приложение отправляет десять сообщений, принимает и подтверждает их, после чего выходит из системы. Для компиляции и упаковки без запуска интеграционного теста используйте mvn -DskipTests package.

Сопоставление управления 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()

Топики JMS и топики служебная шина

Использование тем и подписок служебная шина через JMS API обеспечивает базовые возможности отправки и приёма. Это удобный выбор, когда вы портируете приложения с других брокеров сообщений с API, совместимыми с JMS, хотя разделы служебная шина отличаются от разделов JMS и требуют некоторых корректировок.

Темы служебная шина направляют сообщения в именные, общие и постоянные подписки, которые управляются через интерфейс управления ресурсами Azure, инструменты командной строки Azure или портал Azure. Каждая подписка позволяет использовать до 2000 правил выбора, каждое из которых может иметь условие фильтра, а для SQL-фильтров — также действие преобразования метаданных. Каждое совпадение с условием фильтра выбирает входящее сообщение для копирования в подписку.

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

Темы JMS позволяют клиентам динамически создавать недолговечных и постоянных подписчиков, которые по желанию позволяют фильтровать сообщения с помощью селекторов сообщений. Эти несовместно используемые сущности не поддерживаются стандартом служебная шина. Синтаксис правил SQL-фильтра для служебная шина похож на синтаксис селектора сообщений, поддерживаемый JMS.

Сторона издателя темы JMS совместима со стандартом служебная шина, как показано в примере темы JMS, но динамические подписчики — нет.

Неподдерживаемые функции и ограничения

Следующие ограничения применяются к JMS по AMQP 1.0 с стандартом служебная шина:

  • В одной сессии разрешен только один объект MessageProducer или MessageConsumer. Если вам нужно несколько производителей или потребителей, создайте отдельную сессию для каждого.
  • Подписки на изменяемые темы не поддерживаются.
  • Объекты MessageSelector не поддерживаются.
  • Распределённые транзакции не поддерживаются, но транзакционные сессии поддерживаются.

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

Неподдерживаемый метод Заменить на
createDurableSubscriber Создайте подписку на тему, которая поддерживает селектор сообщений.
createDurableConsumer Создайте подписку на тему с селектором сообщений.
createSharedConsumer Темы служебная шина всегда доступны для общего доступа. См. темы JMS против темы служебная шина.
createSharedDurableConsumer Темы служебная шина всегда доступны для совместного использования. См. темы JMS против темы служебная шина.
создатьВременнуюТему Создайте тему через API управления, инструменты или портал, если для AutoDeleteOnIdle задан период истечения.
createTopic Создайте тему через API управления, инструменты или портал.
Отписаться Удалите тему через API управления, инструменты или портал.
createBrowser Используйте функционал Peek клиентской библиотеки служебная шина.
createQueue Создайте очередь через API управления, инструменты или портал.
createTemporaryQueue Создайте очередь через API управления, инструменты или портал с параметром AutoDeleteOnIdle, установленным на период бездействия.
receiveNoWait Используйте операцию приема с коротким тайм-аутом.

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

  • Последнее обновление: