Отправка сообщений в раздел Служебной шины Azure и получение сообщений из подписок в раздел (Java)

В этом кратком руководстве описано, как написать код Java с помощью пакета azure-messaging-servicebus для отправки сообщений в раздел Служебной шины Azure и их последующего получения из подписок в этот раздел.

Предварительные условия

• Подписка Azure. Для работы с этим учебником требуется учетная запись Azure. Вы можете активировать преимущества подписчика Visual Studio или MSDN или зарегистрироваться для получения бесплатной учетной записи.
• Установите Azure SDK для Java. При использовании Eclipse можно установить набор средств Azure для Eclipse, включающий в себя пакет Azure SDK для Java. Затем можно добавить библиотеки Microsoft Azure для Java в проект. Если вы используете IntelliJ, см. статью Установка набора средств Azure для IntelliJ.

Создание пространства имен на портале Azure

Чтобы начать использование сущностей обмена сообщениями служебной шины в Azure, создайте пространство имен с уникальным именем в Azure. Пространство имен предоставляет контейнер области для ресурсов служебной шины, таких как очереди и разделы в приложении.

Создать пространство имен

1. Войдите на портал Azure.

2. Выберите всплывающее меню в левом верхнем углу и перейдите на страницу "Все службы".

3. На панели навигации слева выберите "Интеграция".

4. Прокрутите вниз дослужебной шины служб >обмена сообщениямии нажмите кнопку "Создать".

   

5. На вкладке "Основы" страницы"Создание пространства имен " выполните следующие действия:

   1. Выберите подписку Azure, в которой будет создано пространство имен.

   2. Для группы ресурсов выберите существующую группу ресурсов или создайте новую.

   3. Введите имя пространства имен, которое соответствует следующим соглашениям об именовании.

      • Это имя должно быть уникальным в пределах Azure. Система немедленно проверяет, доступно ли имя.
      • Длина имени составляет не менее 6 и не более 50 символов.
      • Имя может содержать только буквы, цифры, дефисы -.
      • Имя должно начинаться с буквы и заканчиваться буквой или цифрой.
      • Имя не заканчивается на -sb или -mgmt.

   4. Для расположения выберите регион для размещения пространства имен.

   5. Для параметра Ценовая категория выберите ценовую категорию ("Базовый", "Стандартный" или "Премиум") для пространства имен. Для работы с этим кратким руководством выберите вариант Стандартный.

      При выборе уровня "Премиум" можно включить георепликацию для пространства имен. Функция георепликации гарантирует, что метаданные и данные пространства имен постоянно реплицируются из основного региона в один или несколько дополнительных регионов.

      Внимание

      Чтобы использовать разделы и подписки, выберите категорию "Стандартный" или "Премиум". Темы и подписки не поддерживаются в тарифном уровне "Базовый".

      Если выбрана ценовая категория Премиум, укажите число единиц обмена сообщениями. В категории "Премиум" обеспечивается изоляция ресурсов на уровне ЦП и памяти, так что рабочая нагрузка выполняется изолированно от других. Этот контейнер ресурсов называется единицей обмена сообщениями. Пространству имен ценовой категории "Премиум" выделяется по крайней мере одна единица обмена сообщениями. Для каждого пространства имен служебной шины Premium можно выбрать 1, 2, 4, 8 или 16 единиц обмена сообщениями. Дополнительные сведения см. на уровне обмена сообщениями служебной шины уровня "Премиум".

   6. В нижней части страницы выберите Проверить и создать.

      

   7. На странице Проверить и создать проверьте параметры и нажмите кнопку Создать.

6. После успешного развертывания ресурса выберите "Перейти к ресурсу " на странице развертывания.

   

7. Вы увидите главную страницу пространства имен служебной шины.

   

Создание раздела с помощью портала Azure

1. На странице пространства имен служебной шины разверните сущности в меню навигации слева и выберите темы в меню слева.

2. На панели инструментов выберите + Раздел.

3. Введите имя раздела. Для других параметров оставьте значения по умолчанию.

4. Нажмите кнопку создания.

   

Создание подписки на раздел

1. Выберите раздел, который был создан в предыдущем разделе.

   

2. На странице Тема Служебной шины выберите + Подписка на панели инструментов.

   

3. На странице Создать подписку выполните следующие действия:

   1. Введите S1 в качестве имени подписки.

   2. Выберите Создать, чтобы создать подписку.

      

Проверка подлинности приложения в Azure

В этой статье показано два способа подключения к служебной шине Azure: без пароля и строки подключения.

Первый вариант объясняет, как можно использовать принцип безопасности в Microsoft Entra ID и управление доступом на основе ролей (RBAC) для подключения к пространству имен шины сообщений. Вам не нужно беспокоиться о наличии хардкодированной строки подключения в вашем коде, в файле конфигурации или в безопасном хранилище, таком как Azure Key Vault.

Второй вариант показывает, как использовать строку подключения для подключения к пространству имен Service Bus. Если вы не знакомы с Azure, вы можете найти вариант строка подключения проще следовать. Мы рекомендуем использовать параметр без пароля в реальных приложениях и рабочих средах. Дополнительные сведения см. в разделе "Проверка подлинности и авторизация служебной шины". Дополнительные сведения о проверке подлинности без пароля см. в статье "Проверка подлинности приложений .NET".

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

В этом разделе описано, как создать консольный проект Java и добавить код для отправки сообщений в созданный раздел.

Создание консольного проекта Java

Создайте проект Java с помощью Eclipse или инструмента по своему усмотрению.

Настройка приложения для использования служебной шины

Добавьте ссылку на библиотеки Azure Core и Служебной шины Azure.

Если вы используете Eclipse и создали консольное приложение Java, преобразуйте проект Java в Maven: щелкните проект правой кнопкой мыши в окне обозревателя пакетов, выберите "Настроить -> Maven". Затем добавьте зависимости в эти две библиотеки, как показано в следующем примере.

    <dependencies>
        <dependency>
            <groupId>com.azure</groupId>
            <artifactId>azure-messaging-servicebus</artifactId>
            <version>7.13.3</version>
        </dependency>
        <dependency>
            <groupId>com.azure</groupId>
            <artifactId>azure-identity</artifactId>
            <version>1.8.0</version>
            <scope>compile</scope>
        </dependency>
    </dependencies>

        <dependency>
            <groupId>com.azure</groupId>
            <artifactId>azure-messaging-servicebus</artifactId>
            <version>7.13.3</version>
        </dependency>

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

1. Добавьте в начало Java-файла следующие инструкции import.

   • Без пароля (рекомендуется)
   • Строка подключения

   import com.azure.messaging.servicebus.*;
   import com.azure.identity.*;
   
   import java.util.concurrent.TimeUnit;
   import java.util.Arrays;
   import java.util.List;
   

2. В классе определите переменные для хранения строк подключения (что не требуется для сценария без пароля), имени топика и имени подписки.

   • Без пароля (рекомендуется)
   • Строка подключения

   static String topicName = "<TOPIC NAME>";
   static String subName = "<SUBSCRIPTION NAME>";
   

   Внимание

   Замените <TOPIC NAME> именем раздела и <SUBSCRIPTION NAME> именем подписки раздела.

3. Добавьте метод с именем sendMessage в класс для отправки одного сообщения в раздел.

   • Без пароля (рекомендуется)
   • Строка подключения

   Внимание

   Замените NAMESPACENAME именем пространства имен используемой служебной шины.

   static void sendMessage()
   {
       // create a token using the default Azure credential
       DefaultAzureCredential credential = new DefaultAzureCredentialBuilder()
               .build();
   
       ServiceBusSenderClient senderClient = new ServiceBusClientBuilder()
               .fullyQualifiedNamespace("NAMESPACENAME.servicebus.windows.net")
               .credential(credential)
               .sender()
               .topicName(topicName)
               .buildClient();
   
       // send one message to the topic
       senderClient.sendMessage(new ServiceBusMessage("Hello, World!"));
       System.out.println("Sent a single message to the topic: " + topicName);
   }
   
   

4. Добавьте метод с именем createMessages в класс, чтобы создать список сообщений. Как правило, эти сообщения поступают из различных частей приложения. Здесь мы создадим список примеров сообщений.

   static List<ServiceBusMessage> createMessages()
   {
       // create a list of messages and return it to the caller
       ServiceBusMessage[] messages = {
               new ServiceBusMessage("First message"),
               new ServiceBusMessage("Second message"),
               new ServiceBusMessage("Third message")
       };
       return Arrays.asList(messages);
   }
   

5. Добавьте метод с именем метод sendMessageBatch для отправки сообщений в созданный вами раздел. Этот метод создает ServiceBusSenderClient для раздела, вызывает метод createMessages для получения списка сообщений, а также подготавливает один или несколько пакетов и отправляет пакеты в раздел.

   • Без пароля (рекомендуется)
   • Строка подключения

   Внимание

   Замените NAMESPACENAME именем пространства имен используемой служебной шины.

   static void sendMessageBatch()
   {
       // create a token using the default Azure credential
       DefaultAzureCredential credential = new DefaultAzureCredentialBuilder()
               .build();
   
       ServiceBusSenderClient senderClient = new ServiceBusClientBuilder()
               .fullyQualifiedNamespace("NAMESPACENAME.servicebus.windows.net")
               .credential(credential)
               .sender()
               .topicName(topicName)
               .buildClient();
   
       // Creates an ServiceBusMessageBatch where the ServiceBus.
       ServiceBusMessageBatch messageBatch = senderClient.createMessageBatch();
   
       // create a list of messages
       List<ServiceBusMessage> listOfMessages = createMessages();
   
       // We try to add as many messages as a batch can fit based on the maximum size and send to Service Bus when
       // the batch can hold no more messages. Create a new batch for next set of messages and repeat until all
       // messages are sent.
       for (ServiceBusMessage message : listOfMessages) {
           if (messageBatch.tryAddMessage(message)) {
               continue;
           }
   
           // The batch is full, so we create a new batch and send the batch.
           senderClient.sendMessages(messageBatch);
           System.out.println("Sent a batch of messages to the topic: " + topicName);
   
           // create a new batch
           messageBatch = senderClient.createMessageBatch();
   
           // Add that message that we couldn't before.
           if (!messageBatch.tryAddMessage(message)) {
               System.err.printf("Message is too large for an empty batch. Skipping. Max size: %s.", messageBatch.getMaxSizeInBytes());
           }
       }
   
       if (messageBatch.getCount() > 0) {
           senderClient.sendMessages(messageBatch);
           System.out.println("Sent a batch of messages to the topic: " + topicName);
       }
   
       //close the client
       senderClient.close();
   }
   

Получение сообщений из подписки

В этом разделе вы добавите код для получения сообщений из подписки на тему.

1. Добавьте метод с именем receiveMessages для получения сообщений из подписки. Этот метод создает ServiceBusProcessorClient для подписки, указывая один обработчик для обработки сообщений и другой — для обработки ошибок. Затем он запускает процессор, ждет несколько секунд, выводит полученные сообщения, а затем останавливает и закрывает процессор.

   • Без пароля (рекомендуется)
   • Строка подключения

   Внимание

   • Замените NAMESPACENAME именем пространства имен используемой служебной шины.
   • Замените ServiceBusTopicTest в ServiceBusTopicTest::processMessage в коде именем класса.

   // handles received messages
   static void receiveMessages() throws InterruptedException
   {
       DefaultAzureCredential credential = new DefaultAzureCredentialBuilder()
               .build();
   
       // Create an instance of the processor through the ServiceBusClientBuilder
       ServiceBusProcessorClient processorClient = new ServiceBusClientBuilder()
           .fullyQualifiedNamespace("NAMESPACENAME.servicebus.windows.net")
           .credential(credential)
           .processor()
           .topicName(topicName)
           .subscriptionName(subName)
           .processMessage(context -> processMessage(context))
           .processError(context -> processError(context))
           .buildProcessorClient();
   
       System.out.println("Starting the processor");
       processorClient.start();
   
       TimeUnit.SECONDS.sleep(10);
       System.out.println("Stopping and closing the processor");
       processorClient.close();
   }
   

2. Добавьте метод processMessage для обработки сообщения, полученного из подписки на шину служб.

   private static void processMessage(ServiceBusReceivedMessageContext context) {
       ServiceBusReceivedMessage message = context.getMessage();
       System.out.printf("Processing message. Session: %s, Sequence #: %s. Contents: %s%n", message.getMessageId(),
           message.getSequenceNumber(), message.getBody());
   }
   

3. Добавьте метод processError для обработки сообщений об ошибках.

   private static void processError(ServiceBusErrorContext context) {
       System.out.printf("Error when receiving messages from namespace: '%s'. Entity: '%s'%n",
           context.getFullyQualifiedNamespace(), context.getEntityPath());
   
       if (!(context.getException() instanceof ServiceBusException)) {
           System.out.printf("Non-ServiceBusException occurred: %s%n", context.getException());
           return;
       }
   
       ServiceBusException exception = (ServiceBusException) context.getException();
       ServiceBusFailureReason reason = exception.getReason();
   
       if (reason == ServiceBusFailureReason.MESSAGING_ENTITY_DISABLED
           || reason == ServiceBusFailureReason.MESSAGING_ENTITY_NOT_FOUND
           || reason == ServiceBusFailureReason.UNAUTHORIZED) {
           System.out.printf("An unrecoverable error occurred. Stopping processing with reason %s: %s%n",
               reason, exception.getMessage());
       } else if (reason == ServiceBusFailureReason.MESSAGE_LOCK_LOST) {
           System.out.printf("Message lock lost for message: %s%n", context.getException());
       } else if (reason == ServiceBusFailureReason.SERVICE_BUSY) {
           try {
               // Choosing an arbitrary amount of time to wait until trying again.
               TimeUnit.SECONDS.sleep(1);
           } catch (InterruptedException e) {
               System.err.println("Unable to sleep for period of time");
           }
       } else {
           System.out.printf("Error source %s, reason %s, message: %s%n", context.getErrorSource(),
               reason, context.getException());
       }
   }
   

4. Обновите метод main, чтобы вызвать методы sendMessage, sendMessageBatch и receiveMessages и вызвать InterruptedException.

   public static void main(String[] args) throws InterruptedException {
       sendMessage();
       sendMessageBatch();
       receiveMessages();
   }
   

Выполнить приложение

Выполните программу и убедитесь, что выходные данные выглядят примерно так:

Sent a single message to the topic: mytopic
Sent a batch of messages to the topic: mytopic
Starting the processor
Processing message. Session: e0102f5fbaf646988a2f4b65f7d32385, Sequence #: 1. Contents: Hello, World!
Processing message. Session: 3e991e232ca248f2bc332caa8034bed9, Sequence #: 2. Contents: First message
Processing message. Session: 56d3a9ea7df446f8a2944ee72cca4ea0, Sequence #: 3. Contents: Second message
Processing message. Session: 7bd3bd3e966a40ebbc9b29b082da14bb, Sequence #: 4. Contents: Third message

На странице Обзор для пространства имен служебной шины в портале Azure можно видеть количество входящих и исходящих сообщений. Подождите минуту или около того, а затем обновите страницу, чтобы просмотреть последние значения.

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

Если вы закомментируете вызов receiveMessages в методе main и снова запустите приложение, на странице Раздел служебной шины будет отображаться 8 входящих сообщений (4 новых), и только четыре исходящих.

При выборе подписки на этой странице вы перейдете на страницу подписки на шину сообщений (Service Bus Subscription). Также на этой странице можно увидеть количество активных и отклоненных сообщений, а также многое другое. В этом примере есть четыре активных сообщения, которые получатель еще не получил.

Следующие шаги

Ознакомьтесь со следующими примерами и документацией:

• Клиентская библиотека Служебной шины Azure для Java - файл README
• Примеры на GitHub
• Справочник по API Java