Suporte do Spring Cloud Azure para Spring Integration

2025-06-21
Aplica-se a: ✅ Version 4.20.0, ✅ Version 5.22.0

O Spring Integration Extension for Azure fornece adaptadores Spring Integration para os vários serviços fornecidos pelo SDK do Azure para Java. Fornecemos suporte ao Spring Integration para estes serviços do Azure: Hubs de Eventos, Service Bus, Fila de Armazenamento. Segue-se uma lista de adaptadores suportados:

spring-cloud-azure-starter-integration-eventhubs - para obter mais informações, consulte Integração do Spring com Hubs de Eventos do Azure
spring-cloud-azure-starter-integration-servicebus - para obter mais informações, consulte Integração do Spring com o Azure Service Bus
spring-cloud-azure-starter-integration-storage-queue - para obter mais informações, consulte Integração do Spring com a fila de armazenamento do Azure

Integração do Spring com Hubs de Eventos do Azure

Conceitos-chave

Os Hubs de Eventos do Azure são uma plataforma de streaming de big data e um serviço de ingestão de eventos. Pode receber e processar milhões de eventos por segundo. Os dados enviados para um hub de eventos podem ser transformados e armazenados usando qualquer provedor de análise em tempo real ou adaptadores de lote/armazenamento.

O Spring Integration permite mensagens leves dentro de aplicativos baseados em Spring e suporta a integração com sistemas externos por meio de adaptadores declarativos. Esses adaptadores fornecem um nível mais alto de abstração em relação ao suporte do Spring para comunicação remota, mensagens e agendamento. O projeto de extensão Spring Integration for Event Hubs fornece adaptadores de canal de entrada e saída e gateways para Hubs de Eventos do Azure.

Observação

As APIs de suporte do RxJava são descartadas da versão 4.0.0. Consulte Javadoc para obter detalhes.

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 Kafka armazena todos os deslocamentos confirmados no broker, você precisa armazenar offsets de mensagens de Hubs de Eventos sendo processadas manualmente. O SDK dos Hubs de Eventos fornece a função para armazenar esses deslocamentos dentro do 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 entre consumidores e partições de Kafka, os Hubs de Eventos fornecem uma espécie de modo preventivo. A conta de armazenamento atua como uma concessão para determinar qual partição é de propriedade de qual consumidor. Quando um novo consumidor começa, ele tentará roubar algumas partições da maioria dos consumidores pesados para alcançar o balanceamento da carga de trabalho.

Para especificar a estratégia de balanceamento de carga, os desenvolvedores podem usar EventHubsContainerProperties para a configuração. Consulte seção a seguir para obter um exemplo de como configurar EventHubsContainerProperties.

Suporte ao consumidor em lote

O EventHubsInboundChannelAdapter suporta o modo de consumo em lote. Para habilitá-lo, os usuários podem especificar o modo de ouvinte como ListenerMode.BATCH ao construir uma instância EventHubsInboundChannelAdapter. Quando habilitado, uma mensagem da qual a carga útil é uma lista de eventos em lote será recebida e passada para o canal downstream. Cada cabeçalho de mensagem também é convertido como uma lista, cujo conteúdo é o valor de cabeçalho associado analisado de cada evento. Para os cabeçalhos comuns de ID de partição, ponteiro de verificação e últimas propriedades enfileiradas, eles são apresentados como um único valor para todo o lote de eventos compartilha o mesmo. Para obter mais informações, consulte a seção Cabeçalhos de Mensagem dos Hubs de Eventos.

Observação

O cabeçalho do ponto de verificação só existe quando modo MANUAL ponto de verificação é usado.

O ponto de verificação do consumidor em lote suporta dois modos: BATCH e MANUAL. BATCH modo é um modo de ponto de verificação automático para verificar todo o lote de eventos juntos uma vez que eles são recebidos. MANUAL modo é verificar os eventos pelos usuários. Quando usado, o Checkpointer será passado para o cabeçalho da mensagem e os usuários poderão usá-lo para fazer checkpointing.

A política de consumo em lote pode ser especificada pelas propriedades de max-size e max-wait-time, onde max-size é uma propriedade necessária, enquanto max-wait-time é opcional. Para especificar a estratégia de consumo em lote, os desenvolvedores podem usar EventHubsContainerProperties para a configuração. Consulte seção a seguir para obter um exemplo de como configurar EventHubsContainerProperties.

Configuração de dependência

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

Configuração

Este acionador de partida fornece as seguintes 3 partes das opções de configuração:

Propriedades de configuração da conexão

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

Observação

Se você optar por usar uma entidade de segurança para autenticar e autorizar com a ID do Microsoft Entra para acessar um recurso do Azure, consulte Autorizar acesso com o Microsoft Entra ID para verificar se a entidade de segurança recebeu a permissão suficiente para acessar o recurso do Azure.

Propriedades configuráveis de conexão de spring-cloud-azure-starter-integration-eventhubs:

Propriedade	Tipo	Descrição
spring.cloud.azure.eventhubs.enabled	Booleano	Se um Hubs de Eventos do Azure está habilitado.
spring.cloud.azure.eventhubs.connection-string	Cordão	Valor da cadeia de conexão de namespace de Hubs de Eventos.
spring.cloud.azure.eventhubs.namespace	Cordão	Valor de namespace de Hubs de Eventos, que é o prefixo do FQDN. Um FQDN deve ser composto por NamespaceName.DomainName
spring.cloud.azure.eventhubs.nome-de-domínio	Cordão	Nome de domínio de um valor de Namespace de Hubs de Eventos do Azure.
spring.cloud.azure.eventhubs.custom-endpoint-address	Cordão	Endereço de ponto final personalizado.
spring.cloud.azure.eventhubs.shared-connection	Booleano	Se o EventProcessorClient subjacente e o EventHubProducerAsyncClient usam a mesma conexão. Por padrão, uma nova conexão é construída e usada para cada cliente do Hub de Eventos criado.

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

Esta seção contém as opções de configuração para o serviço de Blobs de Armazenamento, que é usado para manter a propriedade da partição e as informações do ponto de verificação.

Observação

A partir da versão 4.0.0, quando a propriedade de spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not- exists não estiver habilitada manualmente, nenhum contêiner de armazenamento será criado automaticamente.

Propriedades configuráveis de ponto de verificação de spring-cloud-azure-starter-integration-eventhubs:

Propriedade	Tipo	Descrição
spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists	Booleano	Se a criação de contêineres não existe, deve ser permitida.
spring.cloud.azure.eventhubs.processor.checkpoint-store.nome-da-conta	Cordão	Nome da conta de armazenamento.
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-key	Cordão	Chave de acesso da conta de armazenamento.
spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name	Cordão	Nome do recipiente de armazenamento.

As opções comuns de configuração do SDK do Serviço do Azure também são configuráveis para o armazenamento de pontos de verificação de Blob de Armazenamento. As opções de configuração com suporte são introduzidas no de configuração do Spring Cloud Azure e podem ser configuradas com o prefixo unificado spring.cloud.azure. ou o prefixo de spring.cloud.azure.eventhubs.processor.checkpoint-store.

Propriedades de configuração do processador do Hub de Eventos

O EventHubsInboundChannelAdapter usa o EventProcessorClient para consumir mensagens de um hub de eventos, para configurar as propriedades gerais de um EventProcessorClient, os desenvolvedores podem usar EventHubsContainerProperties para a configuração. Consulte seção a seguir sobre como trabalhar com EventHubsInboundChannelAdapter.

Utilização básica

Enviar mensagens para Hubs de Eventos do Azure

Preencha as opções de configuração de credenciais.

Para credenciais como cadeia de conexão, configure as seguintes propriedades no arquivo application.yml:
```
spring:
  cloud:
    azure:
      eventhubs:
        connection-string: ${AZURE_EVENT_HUBS_CONNECTION_STRING}
        processor:
          checkpoint-store:
            container-name: ${CHECKPOINT-CONTAINER}
            account-name: ${CHECKPOINT-STORAGE-ACCOUNT}
            account-key: ${CHECKPOINT-ACCESS-KEY}
```
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 como identidades gerenciadas, configure as seguintes propriedades em seu arquivo application.yml:

spring:
  cloud:
    azure:
      credential:
        managed-identity-enabled: true
        client-id: ${AZURE_CLIENT_ID}
      eventhubs:
        namespace: ${AZURE_EVENT_HUBS_NAMESPACE}
        processor:
          checkpoint-store:
            container-name: ${CONTAINER_NAME}
            account-name: ${ACCOUNT_NAME}

Para credenciais como entidade de serviço, configure as seguintes propriedades no arquivo application.yml:

spring:
  cloud:
    azure:
      credential:
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
      profile:
        tenant-id: <tenant>
      eventhubs:
        namespace: ${AZURE_EVENT_HUBS_NAMESPACE}
        processor:
          checkpoint-store:
            container-name: ${CONTAINER_NAME}
            account-name: ${ACCOUNT_NAME}

Observação

Os valores permitidos para tenant-id são: common, organizations, consumersou o ID do locatário. Para obter mais informações sobre esses valores, consulte a seção Usado o ponto de extremidade errado (contas pessoais e de organização) de AADSTS50020 de erro - A conta de usuário do provedor de identidade não existe nodo locatário . Para obter informações sobre como converter seu aplicativo de locatário único, consulte Converter aplicativo de locatário único em multilocatário no Microsoft Entra ID.

Crie DefaultMessageHandler com o EventHubsTemplate bean para enviar mensagens aos Hubs de Eventos.

class Demo {
    private static final String OUTPUT_CHANNEL = "output";
    private static final String EVENTHUB_NAME = "eh1";

    @Bean
    @ServiceActivator(inputChannel = OUTPUT_CHANNEL)
    public MessageHandler messageSender(EventHubsTemplate eventHubsTemplate) {
        DefaultMessageHandler handler = new DefaultMessageHandler(EVENTHUB_NAME, eventHubsTemplate);
        handler.setSendCallback(new ListenableFutureCallback<Void>() {
            @Override
            public void onSuccess(Void result) {
                LOGGER.info("Message was sent successfully.");
            }
            @Override
            public void onFailure(Throwable ex) {
                LOGGER.error("There was an error sending the message.", ex);
            }
        });
        return handler;
    }
}

Crie uma ligação de gateway de mensagens com o manipulador de mensagens acima por meio de um canal de mensagem.

class Demo {
    @Autowired
    EventHubOutboundGateway messagingGateway;

    @MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
    public interface EventHubOutboundGateway {
        void send(String text);
    }
}

Envie mensagens usando o gateway.

class Demo {
    public void demo() {
        this.messagingGateway.send(message);
    }
}

Receber mensagens dos Hubs de Eventos do Azure

Preencha as opções de configuração de credenciais.

Crie um bean de canal de mensagem como o canal de entrada.

@Configuration
class Demo {
    @Bean
    public MessageChannel input() {
        return new DirectChannel();
    }
}

Crie EventHubsInboundChannelAdapter com o EventHubsMessageListenerContainer bean para receber mensagens de Hubs de Eventos.

@Configuration
class Demo {
    private static final String INPUT_CHANNEL = "input";
    private static final String EVENTHUB_NAME = "eh1";
    private static final String CONSUMER_GROUP = "$Default";

    @Bean
    public EventHubsInboundChannelAdapter messageChannelAdapter(
            @Qualifier(INPUT_CHANNEL) MessageChannel inputChannel,
            EventHubsMessageListenerContainer listenerContainer) {
        EventHubsInboundChannelAdapter adapter = new EventHubsInboundChannelAdapter(processorContainer);
        adapter.setOutputChannel(inputChannel);
        return adapter;
    }

    @Bean
    public EventHubsMessageListenerContainer messageListenerContainer(EventHubsProcessorFactory processorFactory) {
        EventHubsContainerProperties containerProperties = new EventHubsContainerProperties();
        containerProperties.setEventHubName(EVENTHUB_NAME);
        containerProperties.setConsumerGroup(CONSUMER_GROUP);
        containerProperties.setCheckpointConfig(new CheckpointConfig(CheckpointMode.MANUAL));
        return new EventHubsMessageListenerContainer(processorFactory, containerProperties);
    }
}

Crie uma associação de recetor de mensagem com EventHubsInboundChannelAdapter por meio do canal de mensagem criado anteriormente.

class Demo {
    @ServiceActivator(inputChannel = INPUT_CHANNEL)
    public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
        String message = new String(payload);
        LOGGER.info("New message received: '{}'", message);
        checkpointer.success()
                .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message))
                .doOnError(e -> LOGGER.error("Error found", e))
                .block();
    }
}

Configurar o EventHubsMessageConverter para personalizar o objectMapper

EventHubsMessageConverter é feito como um bean configurável para permitir que os usuários personalizem o ObjectMapper.

Suporte ao consumidor em lote

Para consumir mensagens de Hubs de Eventos em lotes é semelhante ao exemplo acima, além disso, os usuários devem definir as opções de configuração relacionadas ao consumo em lote para EventHubsInboundChannelAdapter.

Ao criar EventHubsInboundChannelAdapter, o modo de ouvinte deve ser definido como BATCH. Ao criar o bean de EventHubsMessageListenerContainer, defina o modo de ponto de verificação como MANUAL ou BATCH, e as opções de lote podem ser configuradas conforme necessário.

@Configuration
class Demo {
    private static final String INPUT_CHANNEL = "input";
    private static final String EVENTHUB_NAME = "eh1";
    private static final String CONSUMER_GROUP = "$Default";

    @Bean
    public EventHubsInboundChannelAdapter messageChannelAdapter(
            @Qualifier(INPUT_CHANNEL) MessageChannel inputChannel,
            EventHubsMessageListenerContainer listenerContainer) {
        EventHubsInboundChannelAdapter adapter = new EventHubsInboundChannelAdapter(processorContainer, ListenerMode.BATCH);
        adapter.setOutputChannel(inputChannel);
        return adapter;
    }

    @Bean
    public EventHubsMessageListenerContainer messageListenerContainer(EventHubsProcessorFactory processorFactory) {
        EventHubsContainerProperties containerProperties = new EventHubsContainerProperties();
        containerProperties.setEventHubName(EVENTHUB_NAME);
        containerProperties.setConsumerGroup(CONSUMER_GROUP);
        containerProperties.getBatch().setMaxSize(100);
        containerProperties.setCheckpointConfig(new CheckpointConfig(CheckpointMode.MANUAL));
        return new EventHubsMessageListenerContainer(processorFactory, containerProperties);
    }
}

Cabeçalhos de mensagem dos Hubs de Eventos

A tabela a seguir ilustra como as propriedades de mensagem dos Hubs de Eventos são mapeadas para cabeçalhos de mensagem Spring. Para Hubs de Eventos do Azure, a mensagem é chamada como event.

Mapeamento entre Hubs de Eventos, Propriedades de Mensagem/Evento e Cabeçalhos de Mensagem Spring no Modo de Escuta de Gravação:

Propriedades de eventos de Hubs de Eventos	Constantes de cabeçalho de mensagem de mola	Tipo	Descrição
Tempo enfileirado	EventHubsHeaders#ENQUEUED_TIME	Instantâneo	O instante, em UTC, de quando o evento foi enfileirado na partição do Hub de Eventos.
Deslocamento	EventHubsHeaders#DESLOCAMENTO	Longo	O deslocamento do evento quando ele foi recebido da partição do Hub de Eventos associada.
Chave de partição	AzureHeaders#PARTITION_KEY	Cordão	A chave de hash da partição, se tiver sido definida durante a publicação original do evento.
ID da partição	AzureHeaders#RAW_PARTITION_ID	Cordão	A ID da partição do Hub de Eventos.
Número sequencial	EventHubsHeaders#SEQUENCE_NUMBER	Longo	O número de sequência atribuído ao evento quando ele foi enfileirado na partição do Hub de Eventos associada.
Propriedades do último evento enfileirado	EventHubsHeaders#LAST_ENQUEUED_EVENT_PROPERTIES	LastEnqueuedEventProperties	As propriedades do último evento enfileirado nesta partição.
NA	AzureHeaders#CHECKPOINTER	Ponteiro de verificação	O cabeçalho do ponto de verificação da mensagem específica.

Os usuários podem analisar os cabeçalhos das mensagens para obter as informações relacionadas de cada evento. Para definir um cabeçalho de mensagem para o evento, todos os cabeçalhos personalizados serão colocados como uma propriedade de aplicativo de um evento, onde o cabeçalho é definido como a chave de propriedade. Quando os eventos são recebidos dos Hubs de Eventos, todas as propriedades do aplicativo serão convertidas no cabeçalho da mensagem.

Observação

Não há suporte para cabeçalhos de mensagem de chave de partição, tempo enfileirado, deslocamento e número de sequência para serem definidos manualmente.

Quando o modo batch-consumer está habilitado, os cabeçalhos específicos de mensagens em lote são listados da seguinte forma, que contém uma lista de valores de cada evento de Hubs de Eventos.

Mapeamento entre Hubs de Eventos, Propriedades de Mensagem/Evento e Cabeçalhos de Mensagem primavera no Modo de Escuta em Lote:

Propriedades de eventos de Hubs de Eventos	Constantes de cabeçalho de mensagem de lote de mola	Tipo	Descrição
Tempo enfileirado	EventHubsHeaders#ENQUEUED_TIME	Lista de Instant	Lista do instante, em UTC, de quando cada evento foi enfileirado na partição do Hub de Eventos.
Deslocamento	EventHubsHeaders#DESLOCAMENTO	Lista de Longa	Lista do deslocamento de cada evento quando ele foi recebido da partição do Hub de Eventos associada.
Chave de partição	AzureHeaders#PARTITION_KEY	Lista de String	Lista da chave de hash da partição, se ela foi definida ao publicar originalmente cada evento.
Número sequencial	EventHubsHeaders#SEQUENCE_NUMBER	Lista de Longa	Lista do número de sequência atribuído a cada evento quando ele foi enfileirado na partição do Hub de Eventos associada.
Propriedades do sistema	EventHubsHeaders#BATCH_CONVERTED_SYSTEM_PROPERTIES	Lista de Mapa	Lista das propriedades do sistema de cada evento.
Propriedades do aplicativo	EventHubsHeaders#BATCH_CONVERTED_APPLICATION_PROPERTIES	Lista de Mapa	Lista das propriedades do aplicativo de cada evento, onde todos os cabeçalhos de mensagem personalizados ou propriedades de evento são colocados.

Observação

Ao publicar mensagens, todos os cabeçalhos de lote acima serão removidos das mensagens, se existirem.

Amostras

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

Integração do Spring com o Azure Service Bus

Conceitos-chave

O Spring Integration permite mensagens leves dentro de aplicativos baseados em Spring e suporta a integração com sistemas externos por meio de adaptadores declarativos.

O projeto de extensão Spring Integration for Azure Service Bus fornece adaptadores de canal de entrada e saída para o Azure Service Bus.

Observação

As APIs de suporte do CompletableFuture foram preteridas da versão 2.10.0 e substituídas pelo Reator Core da versão 4.0.0. Consulte Javadoc para obter detalhes.

Configuração de dependência

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

Configuração

Este acionador de partida fornece as seguintes 2 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 ao Barramento de Serviço do Azure.

Observação

Propriedades configuráveis de conexão do spring-cloud-azure-starter-integration-servicebus:

Propriedade	Tipo	Descrição
spring.cloud.azure.servicebus.enabled	Booleano	Se um Barramento de Serviço do Azure está habilitado.
spring.cloud.azure.servicebus.connection-string	Cordão	Valor da cadeia de conexão do Namespace do Service Bus.
spring.cloud.azure.servicebus.custom-endpoint-address	Cordão	O endereço de ponto de extremidade personalizado a ser usado ao se conectar ao Service Bus.
spring.cloud.azure.servicebus.namespace	Cordão	Valor de Namespace do Service Bus, que é o prefixo do FQDN. Um FQDN deve ser composto por NamespaceName.DomainName
spring.cloud.azure.servicebus.nome-de-domínio	Cordão	Nome de domínio de um valor de Namespace do Barramento de Serviço do Azure.

Propriedades de configuração do processador do Service Bus

O ServiceBusInboundChannelAdapter usa o ServiceBusProcessorClient para consumir mensagens, para configurar as propriedades gerais de um ServiceBusProcessorClient, os desenvolvedores podem usar ServiceBusContainerProperties para a configuração. Consulte seção a seguir sobre como trabalhar com ServiceBusInboundChannelAdapter.

Utilização básica

Enviar mensagens para o Barramento de Serviço do Azure

Preencha as opções de configuração de credenciais.
- Para credenciais como cadeia de conexão, configure as seguintes propriedades no arquivo application.yml:
```
spring:
  cloud:
    azure:
      servicebus:
        connection-string: ${AZURE_SERVICE_BUS_CONNECTION_STRING}
```
  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 como identidades gerenciadas, configure as seguintes propriedades em seu arquivo application.yml:
```
spring:
  cloud:
    azure:
      credential:
        managed-identity-enabled: true
        client-id: ${AZURE_CLIENT_ID}
      profile:
        tenant-id: <tenant>
      servicebus:
        namespace: ${AZURE_SERVICE_BUS_NAMESPACE}
```

Observação

Para credenciais como entidade de serviço, configure as seguintes propriedades no arquivo application.yml:

spring:
  cloud:
    azure:
      credential:
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
      profile:
        tenant-id: <tenant>
      servicebus:
        namespace: ${AZURE_SERVICE_BUS_NAMESPACE}

Observação

Crie DefaultMessageHandler com o bean ServiceBusTemplate para enviar mensagens para o Service Bus, defina o tipo de entidade para o ServiceBusTemplate. Este exemplo usa a Fila do Barramento de Serviço como exemplo.

class Demo {
    private static final String OUTPUT_CHANNEL = "queue.output";

    @Bean
    @ServiceActivator(inputChannel = OUTPUT_CHANNEL)
    public MessageHandler queueMessageSender(ServiceBusTemplate serviceBusTemplate) {
        serviceBusTemplate.setDefaultEntityType(ServiceBusEntityType.QUEUE);
        DefaultMessageHandler handler = new DefaultMessageHandler(QUEUE_NAME, serviceBusTemplate);
        handler.setSendCallback(new ListenableFutureCallback<Void>() {
            @Override
            public void onSuccess(Void result) {
                LOGGER.info("Message was sent successfully.");
            }

            @Override
            public void onFailure(Throwable ex) {
                LOGGER.info("There was an error sending the message.");
            }
        });

        return handler;
    }
}

Crie uma ligação de gateway de mensagens com o manipulador de mensagens acima por meio de um canal de mensagem.

class Demo {
    @Autowired
    QueueOutboundGateway messagingGateway;

    @MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
    public interface QueueOutboundGateway {
        void send(String text);
    }
}

Envie mensagens usando o gateway.

class Demo {
    public void demo() {
        this.messagingGateway.send(message);
    }
}

Receber mensagens do Barramento de Serviço do Azure

Preencha as opções de configuração de credenciais.

Crie um bean de canal de mensagem como o canal de entrada.

@Configuration
class Demo {
    private static final String INPUT_CHANNEL = "input";

    @Bean
    public MessageChannel input() {
        return new DirectChannel();
    }
}

Crie ServiceBusInboundChannelAdapter com o ServiceBusMessageListenerContainer bean para receber mensagens no Service Bus. Este exemplo usa a Fila do Barramento de Serviço como exemplo.

@Configuration
class Demo {
    private static final String QUEUE_NAME = "queue1";

    @Bean
    public ServiceBusMessageListenerContainer messageListenerContainer(ServiceBusProcessorFactory processorFactory) {
        ServiceBusContainerProperties containerProperties = new ServiceBusContainerProperties();
        containerProperties.setEntityName(QUEUE_NAME);
        containerProperties.setAutoComplete(false);
        return new ServiceBusMessageListenerContainer(processorFactory, containerProperties);
    }

    @Bean
    public ServiceBusInboundChannelAdapter queueMessageChannelAdapter(
        @Qualifier(INPUT_CHANNEL) MessageChannel inputChannel,
        ServiceBusMessageListenerContainer listenerContainer) {
        ServiceBusInboundChannelAdapter adapter = new ServiceBusInboundChannelAdapter(listenerContainer);
        adapter.setOutputChannel(inputChannel);
        return adapter;
    }
}

Crie uma ligação de recetor de mensagem com ServiceBusInboundChannelAdapter através do canal de mensagens que criamos anteriormente.

class Demo {
    @ServiceActivator(inputChannel = INPUT_CHANNEL)
    public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
        String message = new String(payload);
        LOGGER.info("New message received: '{}'", message);
        checkpointer.success()
                .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message))
                .doOnError(e -> LOGGER.error("Error found", e))
                .block();
    }
}

Configurar ServiceBusMessageConverter para personalizar objectMapper

ServiceBusMessageConverter é feito como um bean configurável para permitir que os usuários personalizem ObjectMapper.

Cabeçalhos de mensagens do Barramento de Serviço

Para alguns cabeçalhos do Service Bus que podem ser mapeados para várias constantes de cabeçalho Spring, a prioridade de cabeçalhos Spring diferentes é listada.

Mapeamento entre cabeçalhos do Service Bus e cabeçalhos de mola:

Cabeçalhos e propriedades de mensagens do Barramento de Serviço	Constantes de cabeçalho de mensagem de mola	Tipo	Configurável	Descrição
Tipo de conteúdo	`MessageHeaders#CONTENT_TYPE`	Cordão	Sim	O descritor RFC2045 Content-Type da mensagem.
ID de correlação	`ServiceBusMessageHeaders#CORRELATION_ID`	Cordão	Sim	O ID de correlação da mensagem
ID da mensagem	`ServiceBusMessageHeaders#MESSAGE_ID`	Cordão	Sim	O ID da mensagem da mensagem, este cabeçalho tem prioridade maior do que `MessageHeaders#ID`.
ID da mensagem	`MessageHeaders#ID`	Identificador Único Universal (UUID)	Sim	O ID da mensagem da mensagem, este cabeçalho tem prioridade menor do que `ServiceBusMessageHeaders#MESSAGE_ID`.
Chave de partição	`ServiceBusMessageHeaders#PARTITION_KEY`	Cordão	Sim	A chave de partição para enviar a mensagem para uma entidade particionada.
Responder a	`MessageHeaders#REPLY_CHANNEL`	Cordão	Sim	O endereço de uma entidade para a qual enviar respostas.
Responder ao ID da sessão	`ServiceBusMessageHeaders#REPLY_TO_SESSION_ID`	Cordão	Sim	O valor da propriedade ReplyToGroupId da mensagem.
Tempo de enfila agendado utc	`ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME`	OffsetDateTime	Sim	A data/hora em que a mensagem deve ser enfileirada no Service Bus, esse cabeçalho tem prioridade maior do que `AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE`.
Tempo de enfila agendado utc	`AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE`	Inteiro	Sim	A data/hora em que a mensagem deve ser enfileirada no Service Bus, esse cabeçalho tem prioridade menor do que `ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME`.
ID da sessão	`ServiceBusMessageHeaders#SESSION_ID`	Cordão	Sim	O IDentifier de sessão para uma entidade com reconhecimento de sessão.
Tempo de viver	`ServiceBusMessageHeaders#TIME_TO_LIVE`	Duração	Sim	A duração do tempo antes que esta mensagem expire.
Para	`ServiceBusMessageHeaders#TO`	Cordão	Sim	O endereço "para" da mensagem, reservado para uso futuro em cenários de roteamento e atualmente ignorado pelo próprio corretor.
Assunto	`ServiceBusMessageHeaders#SUBJECT`	Cordão	Sim	O assunto da mensagem.
Descrição do erro de letra morta	`ServiceBusMessageHeaders#DEAD_LETTER_ERROR_DESCRIPTION`	Cordão	Não	A descrição de uma mensagem com letra morta.
Razão letra morta	`ServiceBusMessageHeaders#DEAD_LETTER_REASON`	Cordão	Não	A razão pela qual uma mensagem estava morta.
Fonte letra morta	`ServiceBusMessageHeaders#DEAD_LETTER_SOURCE`	Cordão	Não	A entidade na qual a mensagem foi escrita sem letra.
Contagem de entregas	`ServiceBusMessageHeaders#DELIVERY_COUNT`	Longo	Não	O número de vezes que esta mensagem foi entregue aos clientes.
Número de sequência enfileirado	`ServiceBusMessageHeaders#ENQUEUED_SEQUENCE_NUMBER`	Longo	Não	O número de sequência enfileirado atribuído a uma mensagem pelo Service Bus.
Tempo enfileirado	`ServiceBusMessageHeaders#ENQUEUED_TIME`	OffsetDateTime	Não	A data/hora em que esta mensagem foi enfileirada no Service Bus.
Expira em	`ServiceBusMessageHeaders#EXPIRES_AT`	OffsetDateTime	Não	A data em que esta mensagem expirará.
Token de bloqueio	`ServiceBusMessageHeaders#LOCK_TOKEN`	Cordão	Não	O token de bloqueio para a mensagem atual.
Bloqueado até	`ServiceBusMessageHeaders#LOCKED_UNTIL`	OffsetDateTime	Não	A data/hora em que o bloqueio desta mensagem expira.
Número sequencial	`ServiceBusMessageHeaders#SEQUENCE_NUMBER`	Longo	Não	O número exclusivo atribuído a uma mensagem pelo Service Bus.
Estado	`ServiceBusMessageHeaders#STATE`	ServiceBusMessageState	Não	O estado da mensagem, que pode ser Ativa, Adiada ou Agendada.

Suporte de chave de partição

Este arranque suporta de particionamento do Service Bus, permitindo definir a chave de partição e o ID da sessão no cabeçalho da mensagem. Esta seção apresenta como definir a chave de partição para mensagens.

Recomendado: Use ServiceBusMessageHeaders.PARTITION_KEY como a chave do cabeçalho.

public class SampleController {
    @PostMapping("/messages")
    public ResponseEntity<String> sendMessage(@RequestParam String message) {
        LOGGER.info("Going to add message {} to Sinks.Many.", message);
        many.emitNext(MessageBuilder.withPayload(message)
                                    .setHeader(ServiceBusMessageHeaders.PARTITION_KEY, "Customize partition key")
                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
        return ResponseEntity.ok("Sent!");
    }
}

Não recomendado, mas atualmente suportado: AzureHeaders.PARTITION_KEY como a chave do cabeçalho.

public class SampleController {
    @PostMapping("/messages")
    public ResponseEntity<String> sendMessage(@RequestParam String message) {
        LOGGER.info("Going to add message {} to Sinks.Many.", message);
        many.emitNext(MessageBuilder.withPayload(message)
                                    .setHeader(AzureHeaders.PARTITION_KEY, "Customize partition key")
                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
        return ResponseEntity.ok("Sent!");
    }
}

Observação

Quando ServiceBusMessageHeaders.PARTITION_KEY e AzureHeaders.PARTITION_KEY são definidos nos cabeçalhos das mensagens, ServiceBusMessageHeaders.PARTITION_KEY é preferível.

Suporte de sessão

Este exemplo demonstra como definir manualmente a ID de sessão de uma mensagem no aplicativo.

public class SampleController {
    @PostMapping("/messages")
    public ResponseEntity<String> sendMessage(@RequestParam String message) {
        LOGGER.info("Going to add message {} to Sinks.Many.", message);
        many.emitNext(MessageBuilder.withPayload(message)
                                    .setHeader(ServiceBusMessageHeaders.SESSION_ID, "Customize session ID")
                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
        return ResponseEntity.ok("Sent!");
    }
}

Observação

Quando o ServiceBusMessageHeaders.SESSION_ID é definido nos cabeçalhos da mensagem e um cabeçalho de ServiceBusMessageHeaders.PARTITION_KEY diferente também é definido, o valor do ID da sessão será eventualmente usado para substituir o valor da chave de partição.

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 obter mais informações, consulte o repositório azure-spring-boot-samples no GitHub.

Integração do Spring com a fila de armazenamento do Azure

Conceitos-chave

O Armazenamento de Filas do Azure é um serviço para armazenar um grande número de mensagens. Você acessa mensagens de qualquer lugar do mundo por meio de chamadas autenticadas usando HTTP ou HTTPS. Uma mensagem de fila pode ter até 64 KB de tamanho. Uma fila pode conter milhões de mensagens, até o limite de capacidade total de uma conta de armazenamento. As filas são comumente usadas para criar uma lista de pendências de trabalho para processar de forma assíncrona.

Configuração de dependência

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter-integration-storage-queue</artifactId>
</dependency>

Configuração

Este acionador de partida fornece as seguintes 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 à Fila de Armazenamento do Azure.

Observação

Propriedades configuráveis de conexão do spring-cloud-azure-starter-integration-storage-queue:

Propriedade	Tipo	Descrição
spring.cloud.azure.storage.queue.enabled	Booleano	Se uma Fila de Armazenamento do Azure está habilitada.
spring.cloud.azure.storage.queue.connection-string	Cordão	Valor da cadeia de conexão do namespace da fila de armazenamento.
spring.cloud.azure.storage.queue.accountName	Cordão	Nome da conta da fila de armazenamento.
spring.cloud.azure.storage.queue.accountKey	Cordão	Chave de conta da fila de armazenamento.
spring.cloud.azure.storage.queue.endpoint	Cordão	Ponto de extremidade do serviço de fila de armazenamento.
spring.cloud.azure.storage.queue.sasToken	Cordão	Credencial de token Sas
spring.cloud.azure.storage.queue.serviceVersão	QueueServiceVersion	QueueServiceVersion que é usado ao fazer solicitações de API.
spring.cloud.azure.storage.queue.messageEncoding	Cordão	Codificação de mensagens em fila.

Utilização básica

Enviar mensagens para a Fila de Armazenamento do Azure

Preencha as opções de configuração de credenciais.
- Para credenciais como cadeia de conexão, configure as seguintes propriedades no arquivo application.yml:
```
spring:
  cloud:
    azure:
      storage:
        queue:
          connection-string: ${AZURE_STORAGE_QUEUE_CONNECTION_STRING}
```
  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 como identidades gerenciadas, configure as seguintes propriedades em seu arquivo application.yml:
```
spring:
  cloud:
    azure:
      credential:
        managed-identity-enabled: true
        client-id: ${AZURE_CLIENT_ID}
      profile:
        tenant-id: <tenant>
      storage:
        queue:
          account-name: ${AZURE_STORAGE_QUEUE_ACCOUNT_NAME}
```

Observação

Para credenciais como entidade de serviço, configure as seguintes propriedades no arquivo application.yml:

spring:
  cloud:
    azure:
      credential:
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
      profile:
        tenant-id: <tenant>
      storage:
        queue:
          account-name: ${AZURE_STORAGE_QUEUE_ACCOUNT_NAME}

Observação

Crie DefaultMessageHandler com o StorageQueueTemplate bean para enviar mensagens para a Fila de Armazenamento.

class Demo {
    private static final String STORAGE_QUEUE_NAME = "example";
    private static final String OUTPUT_CHANNEL = "output";

    @Bean
    @ServiceActivator(inputChannel = OUTPUT_CHANNEL)
    public MessageHandler messageSender(StorageQueueTemplate storageQueueTemplate) {
        DefaultMessageHandler handler = new DefaultMessageHandler(STORAGE_QUEUE_NAME, storageQueueTemplate);
        handler.setSendCallback(new ListenableFutureCallback<Void>() {
            @Override
            public void onSuccess(Void result) {
                LOGGER.info("Message was sent successfully.");
            }

            @Override
            public void onFailure(Throwable ex) {
                LOGGER.info("There was an error sending the message.");
            }
        });
        return handler;
    }
}

Crie uma ligação de gateway de mensagem com o manipulador de mensagens acima por meio de um canal de mensagem.

class Demo {
    @Autowired
    StorageQueueOutboundGateway storageQueueOutboundGateway;

    @MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
    public interface StorageQueueOutboundGateway {
        void send(String text);
    }
}

Envie mensagens usando o gateway.

class Demo {
    public void demo() {
        this.storageQueueOutboundGateway.send(message);
    }
}

Receber mensagens da Fila de Armazenamento do Azure

Preencha as opções de configuração de credenciais.

Crie um bean de canal de mensagem como o canal de entrada.

class Demo {
    private static final String INPUT_CHANNEL = "input";

    @Bean
    public MessageChannel input() {
        return new DirectChannel();
    }
}

Crie StorageQueueMessageSource com o StorageQueueTemplate bean para receber mensagens na Fila de Armazenamento.

class Demo {
    private static final String STORAGE_QUEUE_NAME = "example";

    @Bean
    @InboundChannelAdapter(channel = INPUT_CHANNEL, poller = @Poller(fixedDelay = "1000"))
    public StorageQueueMessageSource storageQueueMessageSource(StorageQueueTemplate storageQueueTemplate) {
        return new StorageQueueMessageSource(STORAGE_QUEUE_NAME, storageQueueTemplate);
    }
}

Crie uma associação de recetor de mensagem com StorageQueueMessageSource criada na última etapa por meio do canal de mensagens que criamos anteriormente.

class Demo {
    @ServiceActivator(inputChannel = INPUT_CHANNEL)
    public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
        String message = new String(payload);
        LOGGER.info("New message received: '{}'", message);
        checkpointer.success()
            .doOnError(Throwable::printStackTrace)
            .doOnSuccess(t -> LOGGER.info("Message '{}' successfully checkpointed", message))
            .block();
    }
}

Amostras

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

Partilhar via

Suporte do Spring Cloud Azure para Spring Integration

Integração do Spring com Hubs de Eventos do Azure

Conceitos-chave

Grupo de consumidores

Suporte de particionamento

Suporte ao consumidor em lote

Configuração de dependência

Configuração

Propriedades de configuração da conexão

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

Propriedades de configuração do processador do Hub de Eventos

Utilização básica

Enviar mensagens para Hubs de Eventos do Azure

Receber mensagens dos Hubs de Eventos do Azure

Configurar o EventHubsMessageConverter para personalizar o objectMapper

Suporte ao consumidor em lote

Cabeçalhos de mensagem dos Hubs de Eventos

Amostras

Integração do Spring com o Azure Service Bus

Conceitos-chave

Configuração de dependência

Configuração

Propriedades de configuração de conexão

Propriedades de configuração do processador do Service Bus

Utilização básica

Enviar mensagens para o Barramento de Serviço do Azure

Receber mensagens do Barramento de Serviço do Azure

Configurar ServiceBusMessageConverter para personalizar objectMapper

Cabeçalhos de mensagens do Barramento de Serviço

Suporte de chave de partição

Suporte de sessão

Personalizar as propriedades do cliente do Service Bus

Amostras

Integração do Spring com a fila de armazenamento do Azure

Conceitos-chave

Configuração de dependência

Configuração

Propriedades de configuração de conexão

Utilização básica

Enviar mensagens para a Fila de Armazenamento do Azure

Receber mensagens da Fila de Armazenamento do Azure

Amostras

Comentários

Recursos adicionais