Отправка и получение сообщений в очередях Azure Service Bus (Java)

В этом кратком руководстве приведены пошаговые инструкции для простого сценария отправки сообщений в очередь Service Bus и их получения. Вы создаете приложение Java для отправки сообщений и получения сообщений из очереди служебной шины Azure. Предварительно созданные примеры Java для служебной шины Azure можно найти в репозитории Azure SDK для Java на сайте GitHub.

Предпосылки

• Подписка Azure. Чтобы завершить это краткое руководство, вам нужна учетная запись Azure. Вы можете активировать ежемесячные кредиты Azure для подписчиков Visual Studio или зарегистрироваться для бесплатной учетной записи.

• Установите 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 единиц обмена сообщениями. Для получения дополнительной информации см. премиальный уровень обмена сообщениями службы Service Bus.

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

      

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

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

   

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

   

Создание очереди на портале Azure

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

2. На странице Очереди, на панели инструментов выберите + Очередь.

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

4. Нажмите кнопку "Создать".

   

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

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

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

Второй вариант показывает, как использовать строку подключения для подключения к пространству имен Service Bus. Если вы не знакомы с Azure, вы можете найти вариант строка подключения проще следовать. Мы рекомендуем использовать параметр без пароля в реальных приложениях и рабочих средах. Дополнительные сведения см. в разделе Проверка подлинности и авторизация Service Bus. Дополнительные сведения о проверке подлинности без пароля см. в статье "Проверка подлинности приложений .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 queueName = "<QUEUE NAME>";
   

   Это важно

   Замените <QUEUE 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()
               .queueName(queueName)
               .buildClient();
   
       // send one message to the queue
       senderClient.sendMessage(new ServiceBusMessage("Hello, World!"));
       System.out.println("Sent a single message to the queue: " + queueName);
   }
   
   

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()
               .queueName(queueName)
               .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 queue: " + queueName);
   
           // 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 queue: " + queueName);
       }
   
       //close the client
       senderClient.close();
   }
   

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

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

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

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

   Это важно

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

   // handles received messages
   static void receiveMessages() throws InterruptedException
   {
       DefaultAzureCredential credential = new DefaultAzureCredentialBuilder()
               .build();
   
       ServiceBusProcessorClient processorClient = new ServiceBusClientBuilder()
               .fullyQualifiedNamespace("NAMESPACENAME.servicebus.windows.net")
               .credential(credential)
               .processor()
               .queueName(queueName)
               .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 queue: myqueue
Sent a batch of messages to the queue: myqueue
Starting the processor
Processing message. Session: 88d961dd801f449e9c3e0f8a5393a527, Sequence #: 1. Contents: Hello, World!
Processing message. Session: e90c8d9039ce403bbe1d0ec7038033a0, Sequence #: 2. Contents: First message
Processing message. Session: 311a216a560c47d184f9831984e6ac1d, Sequence #: 3. Contents: Second message
Processing message. Session: f9a871be07414baf9505f2c3d466c4ab, Sequence #: 4. Contents: Third message
Stopping and closing the processor

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

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

Связанный контент

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

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