Spring Cloud Stream с «Служебная шина Azure»

В этой статье показано, как использовать Spring Cloud Stream Binder для отправки сообщений и получения сообщений от служебная шина queues и topics.

Azure предоставляет асинхронную платформу обмена сообщениями с именем Служебная шина Azure ("служебная шина") на основе стандарта Advanced Message Queueing Protocol 1.0 ("AMQP 1.0"). служебная шина можно использовать в диапазоне поддерживаемых платформ Azure.

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

Примечание.

Чтобы предоставить вашей учетной записи доступ к ресурсам Служебная шина Azure, назначьте роль Служебная шина Azure Data Sender и Служебная шина Azure Data Receiver учетной записи Microsoft Entra, которую вы в данный момент используете. Дополнительные сведения о предоставлении ролей доступа см. в статье Назначение ролей Azure с помощью портала Azure и Аутентификация и авторизация приложения с помощью Microsoft Entra ID для доступа к сущностям службы Служебная шина Azure.

Внимание

Для выполнения действий, описанных в этой статье, требуется spring Boot версии 2.5 или более поздней.

Отправка и получение сообщений от Служебная шина Azure

С помощью очереди или темы для Служебная шина Azure можно отправлять и получать сообщения при помощи Spring Cloud Azure Stream Binder служебная шина.

Чтобы установить модуль Stream Binder Azure служебная шина Spring Cloud, добавьте следующие зависимости в файл pom.xml:

  • Набор спецификаций Spring Cloud Azure (BOM)

    <dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>com.azure.spring</groupId>
      <artifactId>spring-cloud-azure-dependencies</artifactId>
      <version>7.2.0</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>

    Примечание.

    Если вы используете Spring Boot 4.0.x, обязательно установите версию, указанную в spring-cloud-azure-dependencies, на значение 7.2.0.

    Если вы используете Spring Boot 3.5.x, обязательно задайте версию spring-cloud-azure-dependencies на 6.2.0.

    Если вы используете Spring Boot 3.1.x-3.5.x, обязательно установите для нее версию spring-cloud-azure-dependencies на 5.25.0.

    Если вы используете Spring Boot 2.x, обязательно установите версию spring-cloud-azure-dependencies на 4.20.0.

    Эта ведомость материалов (BOM) должна быть сконфигурирована в <dependencyManagement> разделе файла pom.xml. Это гарантирует, что все зависимости Spring Cloud Azure используют одну и ту же версию.

    Дополнительные сведения о версии, используемой для этого BOM, см. в разделе Какую версию Spring Cloud Azure следует использовать.

  • Артефакт Spring Cloud Azure Stream Binder для служебная шина:

    <dependency>
     <groupId>com.azure.spring</groupId>
     <artifactId>spring-cloud-azure-stream-binder-servicebus</artifactId>
</dependency>

Написать код для приложения

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

  1. Настройте учетные данные служебная шина в файле конфигурации application.properties.

     spring.cloud.azure.servicebus.namespace=${AZURE_SERVICEBUS_NAMESPACE}
 spring.cloud.stream.bindings.consume-in-0.destination=${AZURE_SERVICEBUS_QUEUE_NAME}
 spring.cloud.stream.bindings.supply-out-0.destination=${AZURE_SERVICEBUS_QUEUE_NAME}
 spring.cloud.stream.servicebus.bindings.consume-in-0.consumer.auto-complete=false
 spring.cloud.stream.servicebus.bindings.supply-out-0.producer.entity-type=queue
 spring.cloud.function.definition=consume;supply;
 spring.cloud.stream.poller.fixed-delay=60000 
 spring.cloud.stream.poller.initial-delay=0

    В следующей таблице описаны поля в конфигурации:

    Поле Описание
    spring.cloud.azure.servicebus.namespace Укажите пространство имен, полученное в служебная шина на портале Azure.
    spring.cloud.stream.bindings.consume-in-0.destination Укажите очередь служебная шина или тему служебная шина, которую вы использовали в этом руководстве.
    spring.cloud.stream.bindings.supply-out-0.destination Укажите то же значение, которое использовалось для назначения ввода.
    spring.cloud.stream.servicebus.bindings.consume-in-0.consumer.auto-complete Укажите, следует ли автоматически урегулировать сообщения. Если задано значение false, будет добавлен заголовок сообщения Checkpointer, чтобы разработчики могли вручную урегулировать сообщения.
    spring.cloud.stream.servicebus.bindings.supply-out-0.producer.entity-type Укажите тип сущности для выходной привязки, может быть queue или topic.
    spring.cloud.function.definition Укажите, какой функциональный bean-компонент нужно привязать к внешним назначениям, предоставляемым привязками.
    spring.cloud.stream.poller.fixed-delay Укажите фиксированную задержку для опроса по умолчанию в миллисекундах. Значение по умолчанию — 1000 L. Мы рекомендуем использовать значение 60000.
    spring.cloud.stream.poller.initial-delay Укажите начальную задержку для периодических триггеров. Значение по умолчанию — 0.

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

    import com.azure.spring.messaging.checkpoint.Checkpointer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import java.util.function.Consumer;
import java.util.function.Supplier;
import static com.azure.spring.messaging.AzureHeaders.CHECKPOINTER;

@SpringBootApplication
public class ServiceBusQueueBinderApplication implements CommandLineRunner {

    private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusQueueBinderApplication.class);
    private static final Sinks.Many<Message<String>> many = Sinks.many().unicast().onBackpressureBuffer();

    public static void main(String[] args) {
        SpringApplication.run(ServiceBusQueueBinderApplication.class, args);
    }

    @Bean
    public Supplier<Flux<Message<String>>> supply() {
        return ()->many.asFlux()
                       .doOnNext(m->LOGGER.info("Manually sending message {}", m))
                       .doOnError(t->LOGGER.error("Error encountered", t));
    }

    @Bean
    public Consumer<Message<String>> consume() {
        return message->{
            Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
            LOGGER.info("New message received: '{}'", message.getPayload());
            checkpointer.success()
                        .doOnSuccess(s->LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                        .doOnError(e->LOGGER.error("Error found", e))
                        .block();
        };
    }

    @Override
    public void run(String... args) {
        LOGGER.info("Going to add message {} to Sinks.Many.", "Hello World");
        many.emitNext(MessageBuilder.withPayload("Hello World").build(), Sinks.EmitFailureHandler.FAIL_FAST);
    }

}

    Совет

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

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

    Для выполнения проверки подлинности в локальных средах разработки можно использовать Azure CLI, Visual Studio Code, PowerShell или другие методы. Дополнительные сведения см. в разделе Аутентификация в Azure в средах разработки на Java. Чтобы завершить проверку подлинности в средах размещения Azure, рекомендуется использовать управляемое удостоверение, назначаемое пользователем. Дополнительные сведения см. в разделе Что такое управляемые удостоверения для ресурсов Azure

  3. Запустите приложение. Сообщения, как показано в следующем примере, будут размещены в журнале приложений:

    New message received: 'Hello World'
Message 'Hello World' successfully checkpointed

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

Azure для разработчиков SpringSpring Cloud Azure Stream Binder служебная шина Samples

  • Last updated on