Suporte do Spring Cloud Azure para o Spring Cloud Stream

O Spring Cloud Stream é uma estrutura para a criação de microsserviços altamente escaláveis orientados a eventos conectados com sistemas de mensagens compartilhados.

A estrutura fornece um modelo de programação flexível baseado em expressões idiomáticas e melhores práticas Spring já estabelecidas e familiares. Estas práticas recomendadas incluem suporte para semânticas pub/sub persistentes, grupos de consumidores e partições com estado.

As implementações atuais do ligador incluem:

• spring-cloud-azure-stream-binder-eventhubs - para obter mais informações, consulte Spring Cloud Stream Binder for Hubs de Eventos do Azure
• spring-cloud-azure-stream-binder-servicebus - para obter mais informações, consulte Spring Cloud Stream Binder for Azure Service Bus

Spring Cloud Stream Binder para Hubs de Eventos do Azure

Conceitos-chave

O Spring Cloud Stream Binder para Hubs de Eventos do Azure fornece a implementação de vinculação para a estrutura do Spring Cloud Stream. Esta implementação usa os adaptadores de canal do Spring Integration Event Hubs em sua base. Do ponto de vista do design, os Hubs de Eventos são semelhantes a Kafka. Além disso, os Hubs de Eventos podem ser acessados via Kafka API. Se o seu projeto tiver dependência total da API Kafka, você pode tentar Hub de Eventos com a API Kafka Sample

Grupo de consumidores

Os Hubs de Eventos fornecem suporte semelhante ao grupo de consumidores do Apache Kafka, mas com uma lógica ligeiramente diferente. Enquanto o Kafka armazena todos os offsets confirmados no broker, tem de armazenar manualmente os offsets das mensagens do Event Hubs que estão a ser processadas. O SDK do Event Hubs disponibiliza a funcionalidade de armazenar esses offsets no Armazenamento do Azure.

Suporte de particionamento

Os Hubs de Eventos fornecem um conceito de partição física semelhante ao Kafka. Mas, ao contrário do reequilíbrio automático do Kafka entre consumidores e partições, o Event Hubs disponibiliza um tipo de modo preemptivo. A conta de armazenamento atua como uma locação para determinar qual consumidor possui qual partição. Quando um novo consumidor começa, ele tenta roubar algumas partições dos consumidores mais carregados para alcançar o equilíbrio da carga de trabalho.

Para especificar a estratégia de balanceamento de carga, as propriedades de spring.cloud.stream.eventhubs.bindings.<binding-name>.consumer.load-balancing.* são fornecidas. Para obter mais informações, consulte a seção propriedades do consumidor .

Suporte ao consumidor em lote

O binder do Spring Cloud Azure Stream Event Hubs suporta a funcionalidade Spring Cloud Stream Batch Consumer.

Para trabalhar com o modo batch-consumer, defina a propriedade spring.cloud.stream.bindings.<binding-name>.consumer.batch-mode como true. Quando habilitada, uma mensagem com uma carga de uma lista de eventos em lote é recebida e passada para a função Consumer. Cada cabeçalho de mensagem também é convertido em uma lista, cujo conteúdo é o valor de cabeçalho associado analisado de cada evento. Os cabeçalhos comuns de ID de partição, ponteiro de verificação e últimas propriedades enfileiradas são apresentados como um único valor porque todo o lote de eventos compartilha o mesmo valor. Para obter mais informações, consulte a seção cabeçalhos de mensagem dos Hubs de Eventos de suporte do Spring Cloud Azure para o Spring Integration.

O ponto de verificação do consumidor em lote suporta dois modos: BATCH e MANUAL. O modo BATCH é um modo de criação automática de pontos de verificação para criar um ponto de verificação de todo o lote de eventos assim que o binder os recebe. MANUAL modo serve para registar pontos de controlo dos eventos dos utilizadores. Quando utilizado, o Checkpointer é incluído no cabeçalho da mensagem, e os utilizadores podem usá-lo para criar pontos de verificação.

Você pode especificar o tamanho do lote definindo as propriedades max-size e max-wait-time que têm um prefixo de spring.cloud.stream.eventhubs.bindings.<binding-name>.consumer.batch.. A propriedade max-size é necessária e a propriedade max-wait-time é opcional. Para obter mais informações, consulte a seção propriedades do consumidor .

Configuração de dependência

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

Como alternativa, você também pode usar o Spring Cloud Azure Stream Event Hubs Starter, conforme mostrado no exemplo a seguir para o Maven:

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter-stream-eventhubs</artifactId>
</dependency>

Configuração

O fichário fornece as seguintes três partes das opções de configuração:

Propriedades de configuração de conexão

Esta seção contém as opções de configuração usadas para se conectar aos Hubs de Eventos do Azure.

Propriedades configuráveis de ligação de spring-cloud-azure-stream-binder-eventhubs:

O binder também suporta Spring Could Azure Resource Manager por predefinição. Para saber como recuperar a cadeia de ligação com principais de segurança aos quais não foram atribuídas as funções relacionadas com Data, consulte a secção Utilização básica de Spring Could Azure Resource Manager.

Propriedades de configuração do ponto de verificação

Esta secção contém as opções de configuração do serviço Storage Blobs, utilizado para persistir as informações sobre a propriedade das partições e os pontos de controlo.

Pontos de verificação de propriedades configuráveis de spring-cloud-azure-stream-binder-eventhubs:

Propriedades de configuração de vinculação dos Hubs de Eventos do Azure

As opções a seguir estão divididas em quatro seções: Propriedades do consumidor, Configurações avançadas do consumidor, Propriedades do produtor e Configurações avançadas do produtor.

Propriedades do consumidor

Estas propriedades são expostas através de EventHubsConsumerProperties.

Propriedades configuráveis pelo consumidor de spring-cloud-azure-stream-binder-eventhubs:

spring:
  cloud:
    stream:
      eventhubs:
        bindings:
          <binding-name>:
            consumer:
              initial-partition-event-position:
                0:
                  offset: earliest
                1:
                  sequence-number: 100
                2:
                  enqueued-date-time: 2022-01-12T13:32:47.650005Z
                4:
                  inclusive: false

Configuração avançada do cliente

A configuração de ligação, ponto de verificação e cliente comum do SDK do Azure acima suporta personalização para cada consumidor de associador, que pode configurar-se com o prefixo .

Propriedades do produtor

Estas propriedades são expostas através de EventHubsProducerProperties.

Propriedades configuráveis pelo produtor de spring-cloud-azure-stream-binder-eventhubs:

Configuração avançada do produtor

A ligação acima e a configuração do cliente comum do SDK do Azure suportam a personalização para cada produtor de binder, que pode configurar com o prefixo .

Utilização básica

Enviar e receber mensagens de/para Hubs de Eventos

1. Preencha as opções de configuração com informações de credencial.

   • Para credenciais como cadeia de conexão, configure as seguintes propriedades no arquivo application.yml:

     spring:
       cloud:
         azure:
           eventhubs:
             connection-string: ${EVENTHUB_NAMESPACE_CONNECTION_STRING}
             processor:
               checkpoint-store:
                 container-name: ${CHECKPOINT_CONTAINER}
                 account-name: ${CHECKPOINT_STORAGE_ACCOUNT}
                 account-key: ${CHECKPOINT_ACCESS_KEY}
         function:
           definition: consume;supply
         stream:
           bindings:
             consume-in-0:
               destination: ${EVENTHUB_NAME}
               group: ${CONSUMER_GROUP}
             supply-out-0:
               destination: ${THE_SAME_EVENTHUB_NAME_AS_ABOVE}
           eventhubs:
             bindings:
               consume-in-0:
                 consumer:
                   checkpoint:
                     mode: MANUAL
     

     Observação

     A Microsoft recomenda o uso do fluxo de autenticação mais seguro disponível. O fluxo de autenticação descrito neste procedimento, como para bancos de dados, caches, mensagens ou serviços de IA, requer um grau muito alto de confiança no aplicativo e traz riscos não presentes em outros fluxos. Use esse fluxo somente quando opções mais seguras, como identidades gerenciadas para conexões sem senha ou sem chave, não forem viáveis. Para operações de máquina local, prefira identidades de usuário para conexões sem senha ou sem chave.

   • Para credenciais de principal de serviço, configure as seguintes propriedades no ficheiro application.yml:

     spring:
       cloud:
         azure:
           credential:
             client-id: ${AZURE_CLIENT_ID}
             client-secret: ${AZURE_CLIENT_SECRET}
           profile:
             tenant-id: <tenant>
           eventhubs:
             namespace: ${EVENTHUB_NAMESPACE}
             processor:
               checkpoint-store:
                 container-name: ${CONTAINER_NAME}
                 account-name: ${ACCOUNT_NAME}
         function:
           definition: consume;supply
         stream:
           bindings:
             consume-in-0:
               destination: ${EVENTHUB_NAME}
               group: ${CONSUMER_GROUP}
             supply-out-0:
               destination: ${THE_SAME_EVENTHUB_NAME_AS_ABOVE}
           eventhubs:
             bindings:
               consume-in-0:
                 consumer:
                   checkpoint:
                     mode: MANUAL
     

• Para credenciais como identidades gerenciadas, configure as seguintes propriedades em seu arquivo application.yml:

  spring:
    cloud:
      azure:
        credential:
          managed-identity-enabled: true
          client-id: ${AZURE_MANAGED_IDENTITY_CLIENT_ID} # Only needed when using a user-assigned managed identity
        eventhubs:
          namespace: ${EVENTHUB_NAMESPACE}
          processor:
            checkpoint-store:
              container-name: ${CONTAINER_NAME}
              account-name: ${ACCOUNT_NAME}
      function:
        definition: consume;supply
      stream:
        bindings:
          consume-in-0:
            destination: ${EVENTHUB_NAME}
            group: ${CONSUMER_GROUP}
          supply-out-0:
            destination: ${THE_SAME_EVENTHUB_NAME_AS_ABOVE}
  
        eventhubs:
          bindings:
            consume-in-0:
              consumer:
                checkpoint:
                  mode: MANUAL
  

1. Definir fornecedor e consumidor.

   @Bean
   public Consumer<Message<String>> consume() {
       return message -> {
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           LOGGER.info("New message received: '{}', partition key: {}, sequence number: {}, offset: {}, enqueued time: {}",
                   message.getPayload(),
                   message.getHeaders().get(EventHubsHeaders.PARTITION_KEY),
                   message.getHeaders().get(EventHubsHeaders.SEQUENCE_NUMBER),
                   message.getHeaders().get(EventHubsHeaders.OFFSET),
                   message.getHeaders().get(EventHubsHeaders.ENQUEUED_TIME)
           );
   
           checkpointer.success()
                   .doOnSuccess(success -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                   .doOnError(error -> LOGGER.error("Exception found", error))
                   .block();
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply() {
       return () -> {
           LOGGER.info("Sending message, sequence " + i);
           return MessageBuilder.withPayload("Hello world, " + i++).build();
       };
   }
   

Suporte de particionamento

Um PartitionSupplier com informações de partição fornecidas pelo usuário é criado para configurar as informações de partição sobre a mensagem a ser enviada. O fluxograma a seguir mostra o processo de obtenção de diferentes prioridades para o ID e a chave da partição:

Suporte ao consumidor em lote

1. Forneça as opções de configuração de lote, conforme mostrado no exemplo a seguir:

   spring:
     cloud:
       function:
         definition: consume
       stream:
         bindings:
           consume-in-0:
             destination: ${AZURE_EVENTHUB_NAME}
             group: ${AZURE_EVENTHUB_CONSUMER_GROUP}
             consumer:
               batch-mode: true
         eventhubs:
           bindings:
             consume-in-0:
               consumer:
                 batch:
                   max-batch-size: 10 # Required for batch-consumer mode
                   max-wait-time: 1m # Optional, the default value is null
                 checkpoint:
                   mode: BATCH # or MANUAL as needed
   

2. Definir fornecedor e consumidor.

   Para o modo de ponto de verificação como BATCH, você pode usar o código a seguir para enviar mensagens e consumir em lotes.

   @Bean
   public Consumer<Message<List<String>>> consume() {
       return message -> {
           for (int i = 0; i < message.getPayload().size(); i++) {
               LOGGER.info("New message received: '{}', partition key: {}, sequence number: {}, offset: {}, enqueued time: {}",
                       message.getPayload().get(i),
                       ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_PARTITION_KEY)).get(i),
                       ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_SEQUENCE_NUMBER)).get(i),
                       ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_OFFSET)).get(i),
                       ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_ENQUEUED_TIME)).get(i));
           }
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply() {
       return () -> {
           LOGGER.info("Sending message, sequence " + i);
           return MessageBuilder.withPayload("\"test"+ i++ +"\"").build();
       };
   }
   

   Para o modo de checkpointing definido como MANUAL, pode usar o código seguinte para enviar mensagens e consumir/fazer checkpoint em lotes.

   @Bean
   public Consumer<Message<List<String>>> consume() {
       return message -> {
           for (int i = 0; i < message.getPayload().size(); i++) {
               LOGGER.info("New message received: '{}', partition key: {}, sequence number: {}, offset: {}, enqueued time: {}",
                   message.getPayload().get(i),
                   ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_PARTITION_KEY)).get(i),
                   ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_SEQUENCE_NUMBER)).get(i),
                   ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_OFFSET)).get(i),
                   ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_ENQUEUED_TIME)).get(i));
           }
   
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           checkpointer.success()
                       .doOnSuccess(success -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                       .doOnError(error -> LOGGER.error("Exception found", error))
                       .block();
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply() {
       return () -> {
           LOGGER.info("Sending message, sequence " + i);
           return MessageBuilder.withPayload("\"test"+ i++ +"\"").build();
       };
   }
   

Tratar mensagens de erro

Cabeçalhos de mensagem dos Hubs de Eventos

Para obter os cabeçalhos de mensagem básicos suportados, consulte a seção cabeçalhos de mensagem dos Hubs de Eventos de suporte do Spring Cloud Azure para o Spring Integration.

Suporte para vários fichários

A ligação a vários namespaces do Event Hubs também é suportada através da utilização de vários associadores. Este exemplo usa uma cadeia de conexão como exemplo. As credenciais dos principais de serviço e das identidades geridas também são suportadas. Você pode definir propriedades relacionadas nas configurações de ambiente de cada fichário.

1. Para usar vários binders com o Event Hubs, configure as seguintes propriedades no ficheiro application.yml:

   spring:
     cloud:
       function:
         definition: consume1;supply1;consume2;supply2
       stream:
         bindings:
           consume1-in-0:
             destination: ${EVENTHUB_NAME_01}
             group: ${CONSUMER_GROUP_01}
           supply1-out-0:
             destination: ${THE_SAME_EVENTHUB_NAME_01_AS_ABOVE}
           consume2-in-0:
             binder: eventhub-2
             destination: ${EVENTHUB_NAME_02}
             group: ${CONSUMER_GROUP_02}
           supply2-out-0:
             binder: eventhub-2
             destination: ${THE_SAME_EVENTHUB_NAME_02_AS_ABOVE}
         binders:
           eventhub-1:
             type: eventhubs
             default-candidate: true
             environment:
               spring:
                 cloud:
                   azure:
                     eventhubs:
                       connection-string: ${EVENTHUB_NAMESPACE_01_CONNECTION_STRING}
                       processor:
                         checkpoint-store:
                           container-name: ${CHECKPOINT_CONTAINER_01}
                           account-name: ${CHECKPOINT_STORAGE_ACCOUNT}
                           account-key: ${CHECKPOINT_ACCESS_KEY}
           eventhub-2:
             type: eventhubs
             default-candidate: false
             environment:
               spring:
                 cloud:
                   azure:
                     eventhubs:
                       connection-string: ${EVENTHUB_NAMESPACE_02_CONNECTION_STRING}
                       processor:
                         checkpoint-store:
                           container-name: ${CHECKPOINT_CONTAINER_02}
                           account-name: ${CHECKPOINT_STORAGE_ACCOUNT}
                           account-key: ${CHECKPOINT_ACCESS_KEY}
         eventhubs:
           bindings:
             consume1-in-0:
               consumer:
                 checkpoint:
                   mode: MANUAL
             consume2-in-0:
               consumer:
                 checkpoint:
                   mode: MANUAL
         poller:
           initial-delay: 0
           fixed-delay: 1000
   

   Observação

   O ficheiro da aplicação anterior mostra como configurar um único sondador predefinido para ser aplicado a todas as ligações. Se quiser configurar o poller para uma ligação específica, você pode usar uma configuração como spring.cloud.stream.bindings.<binding-name>.producer.poller.fixed-delay=3000.

   Observação

   A Microsoft recomenda o uso do fluxo de autenticação mais seguro disponível. O fluxo de autenticação descrito neste procedimento, como para bancos de dados, caches, mensagens ou serviços de IA, requer um grau muito alto de confiança no aplicativo e traz riscos não presentes em outros fluxos. Use esse fluxo somente quando opções mais seguras, como identidades gerenciadas para conexões sem senha ou sem chave, não forem viáveis. Para operações de máquina local, prefira identidades de usuário para conexões sem senha ou sem chave.

2. Precisamos definir dois fornecedores e dois consumidores:

   @Bean
   public Supplier<Message<String>> supply1() {
       return () -> {
           LOGGER.info("Sending message1, sequence1 " + i);
           return MessageBuilder.withPayload("Hello world1, " + i++).build();
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply2() {
       return () -> {
           LOGGER.info("Sending message2, sequence2 " + j);
           return MessageBuilder.withPayload("Hello world2, " + j++).build();
       };
   }
   
   @Bean
   public Consumer<Message<String>> consume1() {
       return message -> {
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           LOGGER.info("New message1 received: '{}'", message);
           checkpointer.success()
                   .doOnSuccess(success -> LOGGER.info("Message1 '{}' successfully checkpointed", message))
                   .doOnError(error -> LOGGER.error("Exception found", error))
                   .block();
       };
   }
   
   @Bean
   public Consumer<Message<String>> consume2() {
       return message -> {
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           LOGGER.info("New message2 received: '{}'", message);
           checkpointer.success()
                   .doOnSuccess(success -> LOGGER.info("Message2 '{}' successfully checkpointed", message))
                   .doOnError(error -> LOGGER.error("Exception found", error))
                   .block();
       };
   }
   

Provisionamento de recursos

O fichário de Hubs de Eventos oferece suporte ao provisionamento de hub de eventos e grupo de consumidores, os usuários podem usar as seguintes propriedades para habilitar o provisionamento.

spring:
  cloud:
    azure:
      credential:
        tenant-id: <tenant>
      profile:
        subscription-id: ${AZURE_SUBSCRIPTION_ID}
      eventhubs:
        resource:
          resource-group: ${AZURE_EVENTHUBS_RESOURCE_GROUP}

Amostras

Para mais informações, consulte o azure-spring-boot-samples repositório no GitHub.

Spring Cloud Stream Binder para o Azure Service Bus

Conceitos-chave

O Spring Cloud Stream Binder for Azure Service Bus fornece a implementação de vinculação para o Spring Cloud Stream Framework. Esta implementação usa adaptadores de canal do Spring Integration Service Bus em sua base.

Mensagem agendada

Este binder permite submeter mensagens a um tópico para processamento diferido. Os usuários podem enviar mensagens agendadas com cabeçalho x-delay expressando em milissegundos um tempo de atraso para a mensagem. A mensagem será entregue aos respetivos tópicos após x-delay milissegundos.

Grupo de consumidores

O Service Bus Topic fornece suporte semelhante ao grupo de consumidores do Apache Kafka, mas com uma lógica ligeiramente diferente. Este aglutinante baseia-se na Subscription de um tópico para atuar como um grupo de consumidores.

Configuração de dependência

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

Como alternativa, você também pode usar o Spring Cloud Azure Stream Service Bus Starter, conforme mostrado no exemplo a seguir para o Maven:

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

Configuração

O fichário fornece as seguintes duas partes das opções de configuração:

Propriedades de configuração de conexão

Esta secção contém as opções de configuração utilizadas para ligar ao Azure Service Bus.

Propriedades configuráveis de ligação de spring-cloud-azure-stream-binder-servicebus:

O binder também suporta Spring Could Azure Resource Manager por predefinição. Para saber como recuperar a cadeia de ligação com principais de segurança aos quais não foram atribuídas as funções relacionadas com Data, consulte a secção Utilização básica de Spring Could Azure Resource Manager.

Propriedades de configuração da associação do Azure Service Bus

As opções a seguir estão divididas em quatro seções: Propriedades do consumidor, Configurações avançadas do consumidor, Propriedades do produtor e Configurações avançadas do produtor.

Propriedades do consumidor

Estas propriedades são expostas através de ServiceBusConsumerProperties.

Propriedades configuráveis pelo consumidor de spring-cloud-azure-stream-binder-servicebus:

Configuração avançada do utilizador

A configuração de ligação acima e a configuração comum do cliente do SDK do Azure suportam a personalização para cada consumidor do binder, que pode configurar com o prefixo .

Propriedades do produtor

Estas propriedades são expostas através de ServiceBusProducerProperties.

Propriedades configuráveis pelo produtor de spring-cloud-azure-stream-binder-servicebus:

Configuração avançada do produtor

A ligação acima e a configuração do cliente comum do SDK do Azure suportam a personalização para cada produtor de binder, que pode configurar com o prefixo .

Utilização básica

Enviar e receber mensagens de/para o Service Bus

1. Preencha as opções de configuração com informações de credencial.

   • Para credenciais como cadeia de conexão, configure as seguintes propriedades no arquivo application.yml:

     spring:
        cloud:
          azure:
            servicebus:
              connection-string: ${SERVICEBUS_NAMESPACE_CONNECTION_STRING}
          function:
            definition: consume;supply
          stream:
            bindings:
              consume-in-0:
                destination: ${SERVICEBUS_ENTITY_NAME}
                # If you use Service Bus Topic, add the following configuration
                # group: ${SUBSCRIPTION_NAME}
              supply-out-0:
                destination: ${SERVICEBUS_ENTITY_NAME_SAME_AS_ABOVE}
            servicebus:
              bindings:
                consume-in-0:
                  consumer:
                    auto-complete: false
                supply-out-0:
                  producer:
                    entity-type: queue # set as "topic" if you use Service Bus Topic
     

     Observação

     A Microsoft recomenda o uso do fluxo de autenticação mais seguro disponível. O fluxo de autenticação descrito neste procedimento, como para bancos de dados, caches, mensagens ou serviços de IA, requer um grau muito alto de confiança no aplicativo e traz riscos não presentes em outros fluxos. Use esse fluxo somente quando opções mais seguras, como identidades gerenciadas para conexões sem senha ou sem chave, não forem viáveis. Para operações de máquina local, prefira identidades de usuário para conexões sem senha ou sem chave.

   • Para credenciais de principal de serviço, configure as seguintes propriedades no ficheiro application.yml:

     spring:
        cloud:
          azure:
            credential:
              client-id: ${AZURE_CLIENT_ID}
              client-secret: ${AZURE_CLIENT_SECRET}
            profile:
              tenant-id: <tenant>
            servicebus:
              namespace: ${SERVICEBUS_NAMESPACE}
          function:
            definition: consume;supply
          stream:
            bindings:
              consume-in-0:
                destination: ${SERVICEBUS_ENTITY_NAME}
                # If you use Service Bus Topic, add the following configuration
                # group: ${SUBSCRIPTION_NAME}
              supply-out-0:
                destination: ${SERVICEBUS_ENTITY_NAME_SAME_AS_ABOVE}
            servicebus:
              bindings:
                consume-in-0:
                  consumer:
                    auto-complete: false
                supply-out-0:
                  producer:
                    entity-type: queue # set as "topic" if you use Service Bus Topic
     

• Para credenciais como identidades gerenciadas, configure as seguintes propriedades em seu arquivo application.yml:

  spring:
    cloud:
      azure:
        credential:
          managed-identity-enabled: true
          client-id: ${MANAGED_IDENTITY_CLIENT_ID} # Only needed when using a user-assigned managed identity
        servicebus:
          namespace: ${SERVICEBUS_NAMESPACE}
      function:
        definition: consume;supply
      stream:
        bindings:
          consume-in-0:
            destination: ${SERVICEBUS_ENTITY_NAME}
            # If you use Service Bus Topic, add the following configuration
            # group: ${SUBSCRIPTION_NAME}
          supply-out-0:
            destination: ${SERVICEBUS_ENTITY_NAME_SAME_AS_ABOVE}
        servicebus:
          bindings:
            consume-in-0:
              consumer:
                auto-complete: false
            supply-out-0:
              producer:
                entity-type: queue # set as "topic" if you use Service Bus Topic
  

1. Definir fornecedor e consumidor.

   @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(success -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                   .doOnError(error -> LOGGER.error("Exception found", error))
                   .block();
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply() {
       return () -> {
           LOGGER.info("Sending message, sequence " + i);
           return MessageBuilder.withPayload("Hello world, " + i++).build();
       };
   }
   

Suporte de chave de partição

O binder suporta o particionamento do Service Bus ao permitir definir a chave de partição e o ID de sessão no cabeçalho da mensagem. Esta seção apresenta como definir a chave de partição para mensagens.

O Spring Cloud Stream fornece uma propriedade de expressão SpEL para a chave de partição spring.cloud.stream.bindings.<binding-name>.producer.partition-key-expression. Por exemplo, definir essa propriedade como "'partitionKey-' + headers[<message-header-key>]" e adicionar um cabeçalho chamado message-header-key. O Spring Cloud Stream usa o valor desse cabeçalho ao avaliar a expressão para atribuir uma chave de partição. O código a seguir fornece um exemplo de produtor:

@Bean
public Supplier<Message<String>> generate() {
    return () -> {
        String value = "random payload";
        return MessageBuilder.withPayload(value)
            .setHeader("<message-header-key>", value.length() % 4)
            .build();
    };
}

Suporte de sessão

O binder suporta as sessões de mensagens do Service Bus. O ID de sessão de uma mensagem pode ser definido através do cabeçalho da mensagem.

@Bean
public Supplier<Message<String>> generate() {
    return () -> {
        String value = "random payload";
        return MessageBuilder.withPayload(value)
            .setHeader(ServiceBusMessageHeaders.SESSION_ID, "Customize session ID")
            .build();
    };
}

Tratar mensagens de erro

Cabeçalhos de mensagens do Service Bus

Para os cabeçalhos básicos de mensagem suportados, consulte a secção cabeçalhos de mensagem do Service Bus de suporte do Spring Cloud Azure para o Spring Integration.

Suporte para vários fichários

A ligação a vários espaços de nomes do Service Bus também é suportada através da utilização de vários binders. Este exemplo usa a cadeia de conexão como exemplo. Credenciais de entidades de serviço e identidades gerenciadas também são suportadas, os usuários podem definir propriedades relacionadas nas configurações de ambiente de cada fichário.

1. Para usar vários associadores do ServiceBus, configure as seguintes propriedades no ficheiro application.yml:

   spring:
     cloud:
       function:
         definition: consume1;supply1;consume2;supply2
       stream:
         bindings:
           consume1-in-0:
             destination: ${SERVICEBUS_TOPIC_NAME}
             group: ${SUBSCRIPTION_NAME}
           supply1-out-0:
             destination: ${SERVICEBUS_TOPIC_NAME_SAME_AS_ABOVE}
           consume2-in-0:
             binder: servicebus-2
             destination: ${SERVICEBUS_QUEUE_NAME}
           supply2-out-0:
             binder: servicebus-2
             destination: ${SERVICEBUS_QUEUE_NAME_SAME_AS_ABOVE}
         binders:
           servicebus-1:
             type: servicebus
             default-candidate: true
             environment:
               spring:
                 cloud:
                   azure:
                     servicebus:
                       connection-string: ${SERVICEBUS_NAMESPACE_01_CONNECTION_STRING}
           servicebus-2:
             type: servicebus
             default-candidate: false
             environment:
               spring:
                 cloud:
                   azure:
                     servicebus:
                       connection-string: ${SERVICEBUS_NAMESPACE_02_CONNECTION_STRING}
         servicebus:
           bindings:
             consume1-in-0:
               consumer:
                 auto-complete: false
             supply1-out-0:
               producer:
                 entity-type: topic
             consume2-in-0:
               consumer:
                 auto-complete: false
             supply2-out-0:
               producer:
                 entity-type: queue
         poller:
           initial-delay: 0
           fixed-delay: 1000
   

   Observação

   O ficheiro da aplicação anterior mostra como configurar um único sondador predefinido para ser aplicado a todas as ligações. Se quiser configurar o poller para uma ligação específica, você pode usar uma configuração como spring.cloud.stream.bindings.<binding-name>.producer.poller.fixed-delay=3000.

   Observação

   A Microsoft recomenda o uso do fluxo de autenticação mais seguro disponível. O fluxo de autenticação descrito neste procedimento, como para bancos de dados, caches, mensagens ou serviços de IA, requer um grau muito alto de confiança no aplicativo e traz riscos não presentes em outros fluxos. Use esse fluxo somente quando opções mais seguras, como identidades gerenciadas para conexões sem senha ou sem chave, não forem viáveis. Para operações de máquina local, prefira identidades de usuário para conexões sem senha ou sem chave.

2. precisamos definir dois fornecedores e dois consumidores

   @Bean
   public Supplier<Message<String>> supply1() {
       return () -> {
           LOGGER.info("Sending message1, sequence1 " + i);
           return MessageBuilder.withPayload("Hello world1, " + i++).build();
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply2() {
       return () -> {
           LOGGER.info("Sending message2, sequence2 " + j);
           return MessageBuilder.withPayload("Hello world2, " + j++).build();
       };
   }
   
   @Bean
   public Consumer<Message<String>> consume1() {
       return message -> {
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           LOGGER.info("New message1 received: '{}'", message);
           checkpointer.success()
                   .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                   .doOnError(e -> LOGGER.error("Error found", e))
                   .block();
       };
   }
   
   @Bean
   public Consumer<Message<String>> consume2() {
       return message -> {
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           LOGGER.info("New message2 received: '{}'", message);
           checkpointer.success()
                   .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                   .doOnError(e -> LOGGER.error("Error found", e))
                   .block();
       };
   
   }
   

Provisionamento de recursos

O binder do Service Bus suporta o provisionamento de filas, tópicos e subscrições; os utilizadores podem usar as seguintes propriedades para ativar o provisionamento.

spring:
  cloud:
    azure:
      credential:
        tenant-id: <tenant>
      profile:
        subscription-id: ${AZURE_SUBSCRIPTION_ID}
      servicebus:
        resource:
          resource-group: ${AZURE_SERVICEBUS_RESOURCE_GROUP}
    stream:
      servicebus:
        bindings:
          <binding-name>:
            consumer:
              entity-type: ${SERVICEBUS_CONSUMER_ENTITY_TYPE}

Personalizar as propriedades do cliente do Service Bus

Os desenvolvedores podem usar AzureServiceClientBuilderCustomizer para personalizar as propriedades do Service Bus Client. O exemplo a seguir personaliza a propriedade sessionIdleTimeout no ServiceBusClientBuilder:

@Bean
public AzureServiceClientBuilderCustomizer<ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder> customizeBuilder() {
    return builder -> builder.sessionIdleTimeout(Duration.ofSeconds(10));
}

Amostras

Para mais informações, consulte o azure-spring-boot-samples repositório no GitHub.