Soporte de Spring Cloud Azure para Spring Integration

La extensión de integración de Spring para Azure proporciona adaptadores de integración de Spring para los distintos servicios proporcionados por el SDK de Azure para Java. Proporcionamos compatibilidad con Spring Integration para estos servicios de Azure: Event Hubs, Service Bus, Storage Queue. A continuación se muestra una lista de adaptadores admitidos:

• spring-cloud-azure-starter-integration-eventhubs: para más información, consulte Spring Integration con Azure Event Hubs
• spring-cloud-azure-starter-integration-servicebus: para obtener más información, consulte Spring Integration with Azure Service Bus
• spring-cloud-azure-starter-integration-storage-queue: para obtener más información, consulte Spring Integration with Azure Storage Queue

Integración de Spring con Azure Event Hubs

Conceptos clave

Azure Event Hubs es una plataforma de streaming de macrodatos y un servicio de ingesta de eventos. Puede recibir y procesar millones de eventos por segundo. Los datos enviados a un centro de eventos se pueden transformar y almacenar mediante cualquier proveedor de análisis en tiempo real o adaptadores de procesamiento por lotes o almacenamiento.

Spring Integration permite la mensajería ligera dentro de las aplicaciones basadas en Spring y admite la integración con sistemas externos a través de adaptadores declarativos. Esos adaptadores proporcionan un mayor nivel de abstracción sobre el soporte de Spring para el acceso remoto, la mensajería y la planificación de tareas. El proyecto de extensión Spring Integration for Event Hubs proporciona adaptadores de canal entrantes y salientes y puertas de enlace para Azure Event Hubs.

Nota

Las API de compatibilidad con RxJava se eliminan a partir de la versión 4.0.0. Consulte Javadoc para obtener más información.

Grupo de consumidores

Event Hubs proporciona una compatibilidad con grupos de consumidores similar a la de Apache Kafka, aunque con una lógica ligeramente distinta. Aunque Kafka almacena todos los offsets confirmados en el broker, tiene que almacenar manualmente los offsets de los mensajes de Event Hubs que se estén procesando. El SDK de Event Hubs proporciona una función para almacenar estos offsets en Azure Storage.

Compatibilidad con particiones

Event Hubs proporciona un concepto similar de partición física como Kafka. Pero, a diferencia del reequilibrio automático de Kafka entre consumidores y particiones, Event Hubs ofrece una especie de modo preventivo. La cuenta de almacenamiento sirve como mecanismo de arrendamiento para determinar qué consumidor posee cada partición. Cuando se inicie un nuevo consumidor, intentará arrebatar algunas particiones a los consumidores con mayor carga para equilibrar la carga de trabajo.

Para especificar la estrategia de equilibrio de carga, los desarrolladores pueden usar EventHubsContainerProperties para la configuración. Consulte la sección siguiente para obtener un ejemplo de cómo configurar EventHubsContainerProperties.

Compatibilidad con consumidores de lotes

El EventHubsInboundChannelAdapter admite el modo de consumo por lotes. Para habilitarlo, los usuarios pueden especificar el modo de escucha como ListenerMode.BATCH al construir una instancia de EventHubsInboundChannelAdapter. Cuando está habilitado, se recibirá un mensaje cuya carga útil es una lista de eventos agrupados en lotes y se pasará al canal de salida. Cada encabezado de mensaje también se convierte como una lista, de la que el contenido es el valor de encabezado asociado analizado de cada evento. En el caso de los encabezados compartidos de ID de partición, el gestor de puntos de control y las propiedades del último elemento de la cola, se muestran como un único valor si todo el lote de eventos comparte el mismo. Para obtener más información, consulte la sección encabezados de mensajes de Event Hubs.

Nota

El encabezado del punto de control solo existe cuando se utiliza el modo de punto de control MANUAL.

El punto de comprobación del consumidor por lotes admite dos modos: BATCH y MANUAL. El modo BATCH es un modo de creación automática de puntos de control para crear un punto de control de todo el lote de eventos de una sola vez en cuanto se reciben. El modo MANUAL consiste en controlar los eventos por parte de los usuarios. Cuando se use, el Checkpointer se incluirá en la cabecera del mensaje, y los usuarios pueden usarlo para realizar checkpoints.

La directiva de consumo por lotes se puede especificar mediante propiedades de max-size y max-wait-time, donde max-size es una propiedad necesaria mientras max-wait-time es opcional. Para especificar la estrategia de consumo por lotes, los desarrolladores pueden usar EventHubsContainerProperties para la configuración. Consulte la sección siguiente para obtener un ejemplo de cómo configurar EventHubsContainerProperties.

Configuración de dependencias

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

Configuración

Este inicio proporciona las siguientes tres partes de las opciones de configuración:

Propiedades de configuración de conexión

Esta sección contiene las opciones de configuración que se usan para conectarse a Azure Event Hubs.

Nota

Si decide usar una entidad de seguridad para autenticar y autorizar con el identificador de Entra de Microsoft para acceder a un recurso de Azure, consulte Autorizar el acceso con el identificador de Entra de Microsoft para asegurarse de que a la entidad de seguridad se le ha concedido el permiso suficiente para acceder al recurso de Azure.

Propiedades configurables de conexión de spring-cloud-azure-starter-integration-eventhubs:

Propiedad	Tipo	Descripción
spring.cloud.azure.eventhubs.enabled	booleano	Si una instancia de Azure Event Hubs está habilitada.
spring.cloud.azure.eventhubs.connection-string	Cuerda	Valor de cadena de conexión del espacio de nombres de Event Hubs.
spring.cloud.azure.eventhubs.namespace	Cuerda	Valor del espacio de nombres de Event Hubs, que es el prefijo del FQDN. Un FQDN debe estar compuesto por NamespaceName.DomainName
spring.cloud.azure.eventhubs.domain-name	Cuerda	Valor del nombre de dominio de un espacio de nombres de Azure Event Hubs.
spring.cloud.azure.eventhubs.custom-endpoint-address	Cuerda	Dirección del punto de conexión personalizado.
spring.cloud.azure.eventhubs.shared-connection	Booleano	Si EventProcessorClient y EventHubProducerAsyncClient subyacente usan la misma conexión. De forma predeterminada, se crea una nueva conexión y se usa para cada cliente del centro de eventos creado.

Propiedades de configuración del punto de control

Esta sección contiene las opciones de configuración del servicio Storage Blobs, que se usa para conservar la propiedad de la partición y la información del punto de control.

Nota

Desde la versión 4.0.0, cuando la propiedad de spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists no está habilitada manualmente, no se creará automáticamente ningún contenedor de almacenamiento.

Propiedades configurables del punto de control de spring-cloud-azure-starter-integration-eventhubs:

Propiedad	Tipo	Descripción
spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists	Booleano	Si se permite crear contenedores si no existen.
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name	Cuerda	Nombre de la cuenta de almacenamiento.
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-key	Cuerda	Clave de acceso de la cuenta de almacenamiento.
spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name	Cuerda	Nombre del contenedor de almacenamiento.

Las opciones de configuración comunes del SDK de Azure Service también pueden configurarse para el almacén de puntos de control de Storage Blob. Las opciones de configuración admitidas se describen en Configuración de Spring Cloud Azure y se pueden configurar con el prefijo unificado spring.cloud.azure. o con el prefijo spring.cloud.azure.eventhubs.processor.checkpoint-store..

Propiedades de configuración del procesador del centro de eventos

El EventHubsInboundChannelAdapter usa el EventProcessorClient para consumir mensajes de un centro de eventos, para configurar las propiedades generales de un EventProcessorClient, los desarrolladores pueden usar EventHubsContainerProperties para la configuración. Consulte la sección siguiente sobre cómo trabajar con EventHubsInboundChannelAdapter.

Uso básico

Envío de mensajes a Azure Event Hubs

1. Rellene las opciones de configuración de credenciales.

   • Para las credenciales como cadena de conexión, configure las siguientes propiedades en el archivo 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}
     

     Nota

     Microsoft recomienda usar el flujo de autenticación más seguro disponible. El flujo de autenticación descrito en este procedimiento, como para bases de datos, memorias caché, mensajería o servicios de inteligencia artificial, requiere un grado de confianza muy alto en la aplicación y conlleva riesgos que no están presentes en otros flujos. Use este flujo solo cuando las opciones más seguras, como las identidades administradas para conexiones sin contraseña o sin claves, no sean viables. En el caso de las operaciones de máquina local, prefiera identidades de usuario para conexiones sin contraseña o sin claves.

   • Para las credenciales como identidades administradas, configure las siguientes propiedades en el archivo 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 las credenciales como principal de servicio, configure las siguientes propiedades en su archivo 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}
     

Nota

Los valores permitidos para tenant-id son: common, organizations, consumerso el identificador de inquilino. Para obtener más información sobre estos valores, consulte la sección Uso del punto de conexión incorrecto (cuentas personales y de organización) de Error AADSTS50020: la cuenta de usuario del proveedor de identidades no existe en el inquilino. Para obtener información sobre cómo convertir la aplicación de un solo inquilino, consulte Convertir aplicación de un solo inquilino en multiinquilino en microsoft Entra ID.

1. Cree DefaultMessageHandler con el bean EventHubsTemplate para enviar mensajes a Event Hubs.

   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;
       }
   }
   

2. Cree una vinculación de puerta de enlace de mensajes con el controlador de mensajes anterior a través de un canal de mensajes.

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

3. Enviar mensajes mediante la puerta de enlace.

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

Recepción de mensajes de Azure Event Hubs

1. Rellene las opciones de configuración de credenciales.

2. Cree un bean del canal de mensajes como canal de entrada.

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

3. Crea EventHubsInboundChannelAdapter con el componente EventHubsMessageListenerContainer para recibir mensajes de Event Hubs.

   @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(listenerContainer);
           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);
       }
   }
   

4. Cree un enlace de receptor de mensajes con EventHubsInboundChannelAdapter a través del canal de mensajes creado antes.

   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();
       }
   }
   

Configuración de EventHubsMessageConverter para personalizar objectMapper

EventHubsMessageConverter se realiza como un bean configurable para permitir a los usuarios personalizar ObjectMapper.

Compatibilidad con consumidores de lotes

Consumir mensajes de Event Hubs por lotes es similar al ejemplo anterior, pero además los usuarios deben establecer las opciones de configuración relacionadas con el consumo por lotes para EventHubsInboundChannelAdapter.

Al crear EventHubsInboundChannelAdapter, el modo de escucha debe establecerse como BATCH. Cuando cree el bean de EventHubsMessageListenerContainer, establezca el modo de punto de control en MANUAL o BATCH. Además, puede configurar las opciones de lote según sea necesario.

@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(listenerContainer, 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);
    }
}

Encabezados de mensaje de Event Hubs

En la tabla siguiente se muestra cómo se asignan las propiedades del mensaje de Event Hubs a los encabezados de mensaje de Spring. Para Azure Event Hubs, se llama al mensaje como event.

Correspondencia entre las propiedades de mensaje y evento de Event Hubs y los encabezados de mensaje de Spring en modo de escucha de registros:

Propiedades del evento de Event Hubs	Constantes de encabezado de mensaje de Spring	Tipo	Descripción
Hora de puesta en cola	EventHubsHeaders#ENQUEUED_TIME	Instantáneo	El instante, en UTC, en que el evento se encoló en la partición de Event Hub.
Compensar	EventHubsHeaders#OFFSET	Largo	El desfase del evento en el momento en que se recibió de la partición de Event Hub asociada.
Clave de partición	AzureHeaders#PARTITION_KEY	Cuerda	La clave de hash de la partición, si se estableció cuando se publicó originalmente el evento.
Id. de partición	AzureHeaders#RAW_PARTITION_ID	Cuerda	El identificador de la partición del Event Hub.
Número de secuencia	EventHubsHeaders#SEQUENCE_NUMBER	Largo	El número de secuencia asignado al evento cuando se encoló en la partición del Event Hub asociado.
Propiedades del último evento puesto en cola	EventHubsHeaders#LAST_ENQUEUED_EVENT_PROPERTIES	LastEnqueuedEventProperties	Propiedades del último evento en cola en esta partición.
NA	AzureHeaders#CHECKPOINTER	Gestor de puntos de control	Encabezado del punto de control para el mensaje específico.

Los usuarios pueden analizar los encabezados de mensaje para obtener la información relacionada de cada evento. Para establecer un encabezado de mensaje para el evento, todos los encabezados personalizados se colocarán como una propiedad de aplicación de un evento, donde el encabezado se establece como clave de propiedad. Cuando se reciben eventos de Event Hubs, todas las propiedades de la aplicación se convertirán en el encabezado del mensaje.

Nota

No se admite establecer manualmente los encabezados del mensaje correspondientes a la clave de partición, la hora de puesta en cola, el desplazamiento y el número de secuencia.

Cuando el modo de consumidor por lotes está habilitado, los encabezados específicos de los mensajes en lote son los siguientes y contienen una lista de valores de cada evento individual de Event Hubs.

Asignación entre las propiedades de mensaje y de evento de Event Hubs y los encabezados de mensaje de Spring en el modo de escucha por lotes:

Propiedades del evento de Event Hubs	Constantes de encabezado de mensaje de Spring Batch	Tipo	Descripción
Hora de puesta en cola	EventHubsHeaders#ENQUEUED_TIME	Lista de instantáneas	Lista de los instantes, en UTC, en que se encoló cada evento en la partición de Event Hub.
Compensar	EventHubsHeaders#OFFSET	Lista larga	Lista de desplazamientos de cada evento en el momento en que se recibió de la partición de Event Hub asociada.
Clave de partición	AzureHeaders#PARTITION_KEY	Lista de cadenas	Lista de la clave hash de la partición, si se estableció cuando se publicó originalmente cada evento.
Número de secuencia	EventHubsHeaders#SEQUENCE_NUMBER	Lista larga	Lista de los números de secuencia asignados a cada evento cuando se encoló en la partición de Event Hub correspondiente.
Propiedades del sistema	EventHubsHeaders#BATCH_CONVERTED_SYSTEM_PROPERTIES	Lista de mapas	Lista de las propiedades del sistema de cada evento.
Propiedades de la aplicación	EventHubsHeaders#BATCH_CONVERTED_APPLICATION_PROPERTIES	Lista de mapas	Lista de las propiedades de la aplicación de cada evento, donde se colocan todos los encabezados de mensaje personalizados o propiedades de evento.

Nota

Al publicar mensajes, todas las cabeceras de lote indicadas anteriormente se eliminarán de los mensajes si están presentes.

Muestras

Para obtener más información, consulte el azure-spring-boot-samples repositorio en GitHub.

Integración de Spring con Azure Service Bus

Conceptos clave

Spring Integration permite la mensajería ligera dentro de las aplicaciones basadas en Spring y admite la integración con sistemas externos a través de adaptadores declarativos.

El proyecto de extensión Spring Integration for Azure Service Bus proporciona adaptadores de canal entrantes y salientes para Azure Service Bus.

Nota

Las API compatibles con CompletableFuture se han marcado como obsoletas a partir de la versión 2.10.0 y se han sustituido por Reactor Core a partir de la versión 4.0.0. Consulte Javadoc para obtener más información.

Configuración de dependencias

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

Configuración

Este inicio proporciona las siguientes 2 partes de las opciones de configuración:

Propiedades de configuración de conexión

Esta sección contiene las opciones de configuración que se usan para conectarse a Azure Service Bus.

Nota

Si decide usar una entidad de seguridad para autenticar y autorizar con el identificador de Entra de Microsoft para acceder a un recurso de Azure, consulte Autorizar el acceso con el identificador de Entra de Microsoft para asegurarse de que a la entidad de seguridad se le ha concedido el permiso suficiente para acceder al recurso de Azure.

Propiedades configurables de conexión de spring-cloud-azure-starter-integration-servicebus:

Propiedad	Tipo	Descripción
spring.cloud.azure.servicebus.enabled	booleano	Indica si una instancia de Azure Service Bus está habilitada.
spring.cloud.azure.servicebus.connection-string	Cuerda	Valor de la cadena de conexión del espacio de nombres de Service Bus.
spring.cloud.azure.servicebus.custom-endpoint-address	Cuerda	Dirección del punto de conexión personalizado que se va a usar al conectarse a Service Bus.
spring.cloud.azure.servicebus.namespace	Cuerda	Valor del espacio de nombres de Service Bus, que es el prefijo del FQDN. Un FQDN debe estar compuesto por NamespaceName.DomainName
spring.cloud.azure.servicebus.domain-name	Cuerda	Nombre de dominio de un valor de espacio de nombres de Azure Service Bus.

Propiedades de configuración del procesador de Service Bus

El ServiceBusInboundChannelAdapter usa el ServiceBusProcessorClient para consumir mensajes, para configurar las propiedades generales de un ServiceBusProcessorClient, los desarrolladores pueden usar ServiceBusContainerProperties para la configuración. Consulte la sección siguiente sobre cómo trabajar con ServiceBusInboundChannelAdapter.

Uso básico

Envío de mensajes a Azure Service Bus

1. Rellene las opciones de configuración de credenciales.

   • Para las credenciales como cadena de conexión, configure las siguientes propiedades en el archivo application.yml:

     spring:
       cloud:
         azure:
           servicebus:
             connection-string: ${AZURE_SERVICE_BUS_CONNECTION_STRING}
     

     Nota

     Microsoft recomienda usar el flujo de autenticación más seguro disponible. El flujo de autenticación descrito en este procedimiento, como para bases de datos, memorias caché, mensajería o servicios de inteligencia artificial, requiere un grado de confianza muy alto en la aplicación y conlleva riesgos que no están presentes en otros flujos. Use este flujo solo cuando las opciones más seguras, como las identidades administradas para conexiones sin contraseña o sin claves, no sean viables. En el caso de las operaciones de máquina local, prefiera identidades de usuario para conexiones sin contraseña o sin claves.

   • Para las credenciales como identidades administradas, configure las siguientes propiedades en el archivo 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}
     

Nota

Los valores permitidos para tenant-id son: common, organizations, consumerso el identificador de inquilino. Para obtener más información sobre estos valores, consulte la sección Uso del punto de conexión incorrecto (cuentas personales y de organización) de Error AADSTS50020: la cuenta de usuario del proveedor de identidades no existe en el inquilino. Para obtener información sobre cómo convertir la aplicación de un solo inquilino, consulte Convertir aplicación de un solo inquilino en multiinquilino en microsoft Entra ID.

• Para las credenciales como principal de servicio, configure las siguientes propiedades en su archivo 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}
  

Nota

Los valores permitidos para tenant-id son: common, organizations, consumerso el identificador de inquilino. Para obtener más información sobre estos valores, consulte la sección Uso del punto de conexión incorrecto (cuentas personales y de organización) de Error AADSTS50020: la cuenta de usuario del proveedor de identidades no existe en el inquilino. Para obtener información sobre cómo convertir la aplicación de un solo inquilino, consulte Convertir aplicación de un solo inquilino en multiinquilino en microsoft Entra ID.

1. Cree DefaultMessageHandler con el bean ServiceBusTemplate para enviar mensajes a Service Bus y establezca el tipo de entidad para ServiceBusTemplate. En este ejemplo se toma la cola de Service Bus como ejemplo.

   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.error("There was an error sending the message.", ex);
               }
           });
   
           return handler;
       }
   }
   

2. Cree una vinculación de puerta de enlace de mensajes con el controlador de mensajes anterior a través de un canal de mensajes.

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

3. Enviar mensajes mediante la puerta de enlace.

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

Recepción de mensajes de Azure Service Bus

1. Rellene las opciones de configuración de credenciales.

2. Cree un bean del canal de mensajes como canal de entrada.

   @Configuration
   class Demo {
       private static final String INPUT_CHANNEL = "input";
   
       @Bean
       public MessageChannel input() {
           return new DirectChannel();
       }
   }
   

3. Cree ServiceBusInboundChannelAdapter con el bean ServiceBusMessageListenerContainer para recibir mensajes de Service Bus. En este ejemplo se toma la cola de Service Bus como ejemplo.

   @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;
       }
   }
   

4. Cree un enlace de receptor de mensajes con ServiceBusInboundChannelAdapter a través del canal de mensajes que creamos antes.

   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();
       }
   }
   

Configuración de ServiceBusMessageConverter para personalizar objectMapper

ServiceBusMessageConverter se realiza como un bean configurable para permitir a los usuarios personalizar ObjectMapper.

Encabezados de mensaje de Service Bus

Para algunos encabezados de Service Bus que se pueden asignar a varias constantes de encabezado de Spring, se muestra la prioridad de distintos encabezados de Spring.

Correspondencia entre encabezados de Service Bus y encabezados de Spring:

Cabeceras y propiedades del mensaje de Service Bus	Constantes de encabezado del mensaje de Spring	Tipo	Configurable	Descripción
Tipo de contenido	MessageHeaders#CONTENT_TYPE	Cuerda	Sí	Descriptor de tipo de contenido RFC2045 del mensaje.
Id. de correlación	ServiceBusMessageHeaders#CORRELATION_ID	Cuerda	Sí	Identificador de correlación del mensaje
Id. de mensaje	ServiceBusMessageHeaders#MESSAGE_ID	Cuerda	Sí	El identificador de mensaje del mensaje, este encabezado tiene mayor prioridad que MessageHeaders#ID.
Id. de mensaje	MessageHeaders#ID	Identificador Único Universal (UUID)	Sí	El identificador de mensaje del mensaje, este encabezado tiene una prioridad menor que ServiceBusMessageHeaders#MESSAGE_ID.
Clave de partición	ServiceBusMessageHeaders#PARTITION_KEY	Cuerda	Sí	Clave de partición para enviar el mensaje a una entidad con particiones.
Responder a	MessageHeaders#REPLY_CHANNEL	Cuerda	Sí	Dirección de una entidad a la que se van a enviar respuestas.
Responder al identificador de sesión	ServiceBusMessageHeaders#REPLY_TO_SESSION_ID	Cuerda	Sí	El valor de la propiedad ReplyToGroupId del mensaje.
Hora programada de entrada en cola (UTC)	ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME	OffsetDateTime	Sí	La fecha y hora en que se debe encolar el mensaje en el Service Bus; esta cabecera tiene mayor prioridad que AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE.
Hora programada de puesta en cola UTC	AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE	Entero	Sí	La fecha y hora en que debe ponerse en cola el mensaje en Service Bus; esta cabecera tiene menos prioridad que ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME.
Identificador de sesión	ServiceBusMessageHeaders#SESSION_ID	Cuerda	Sí	El identificador de sesión de una entidad con reconocimiento de sesión.
Tiempo de vida	ServiceBusMessageHeaders#TIME_TO_LIVE	Duración	Sí	Duración del tiempo antes de que expire este mensaje.
Para	ServiceBusMessageHeaders#TO	Cuerda	Sí	La dirección "to" del mensaje, reservada para usos futuros en escenarios de enrutamiento y actualmente ignorada por el propio intermediario.
Asunto	ServiceBusMessageHeaders#SUBJECT	Cuerda	Sí	Asunto del mensaje.
Descripción del error de mensaje fallido	ServiceBusMessageHeaders#DEAD_LETTER_ERROR_DESCRIPTION	Cuerda	No	Descripción de un mensaje fallido.
Motivo del mensaje fallido	ServiceBusMessageHeaders#DEAD_LETTER_REASON	Cuerda	No	Razón por la que se ha generado un mensaje fallido.
Origen del mensaje fallido	ServiceBusMessageHeaders#DEAD_LETTER_SOURCE	Cuerda	No	Entidad donde se encontraba el mensaje fallido.
Recuento de entregas	ServiceBusMessageHeaders#DELIVERY_COUNT	largo	No	Número de veces que este mensaje se entregó a los clientes.
Número de secuencia en cola	ServiceBusMessageHeaders#ENQUEUED_SEQUENCE_NUMBER	largo	No	Número secuencial en cola asignado a un mensaje por Service Bus.
Hora de puesta en cola	ServiceBusMessageHeaders#ENQUEUED_TIME	OffsetDateTime	No	Fecha y hora en la que se puso en cola este mensaje en Service Bus.
Expira en	ServiceBusMessageHeaders#EXPIRES_AT	OffsetDateTime	No	Fecha y hora en la que expirará este mensaje.
Token de bloqueo	ServiceBusMessageHeaders#LOCK_TOKEN	Cuerda	No	Token de bloqueo del mensaje actual.
Bloqueado hasta	ServiceBusMessageHeaders#LOCKED_UNTIL	OffsetDateTime	No	Fecha y hora en la que expira el bloqueo de este mensaje.
Número de secuencia	ServiceBusMessageHeaders#SEQUENCE_NUMBER	largo	No	Número único asignado a un mensaje por Service Bus.
Estado	ServiceBusMessageHeaders#STATE	ServiceBusMessageState	No	Estado del mensaje, que puede ser Activo, Diferido o Programado.

Compatibilidad con claves de partición

Este inicio admite la creación de particiones de Service Bus, ya que permite establecer la clave de partición y el identificador de sesión en el encabezado del mensaje. En esta sección se presenta cómo establecer la clave de partición para los mensajes.

Recomendado: use ServiceBusMessageHeaders.PARTITION_KEY como clave del encabezado.

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!");
    }
}

No se recomienda, pero actualmente se admite: AzureHeaders.PARTITION_KEY como clave del encabezado.

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!");
    }
}

Nota

Cuando ServiceBusMessageHeaders.PARTITION_KEY y AzureHeaders.PARTITION_KEY se establecen en los encabezados del mensaje, se prefiere ServiceBusMessageHeaders.PARTITION_KEY.

Soporte de sesiones

En este ejemplo se muestra cómo establecer manualmente el identificador de sesión de un mensaje en la aplicación.

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!");
    }
}

Nota

Cuando el ServiceBusMessageHeaders.SESSION_ID se establece en los encabezados del mensaje y también se establece un encabezado de ServiceBusMessageHeaders.PARTITION_KEY diferente, el valor del identificador de sesión se usará finalmente para sobrescribir el valor de la clave de partición.

Personalización de las propiedades del cliente de Service Bus

Los desarrolladores pueden usar AzureServiceClientBuilderCustomizer para personalizar las propiedades del cliente de Service Bus. En el ejemplo siguiente se personaliza la propiedad sessionIdleTimeout en ServiceBusClientBuilder:

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

Muestras

Para obtener más información, consulte el azure-spring-boot-samples repositorio en GitHub.

Integración de Spring con Azure Storage Queue

Conceptos clave

Azure Queue Storage es un servicio para almacenar un gran número de mensajes. Puede acceder a los mensajes desde cualquier lugar del mundo a través de llamadas autenticadas mediante HTTP o HTTPS. Un mensaje de cola puede tener un tamaño de hasta 64 KB. Una cola puede contener millones de mensajes, hasta el límite de capacidad total de una cuenta de almacenamiento. Las colas se usan normalmente para crear un trabajo pendiente para procesar de forma asincrónica.

Configuración de dependencias

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

Configuración

Este inicio proporciona las siguientes opciones de configuración:

Propiedades de configuración de conexión

Esta sección contiene las opciones de configuración que se usan para conectarse a la cola de Azure Storage.

Nota

Si decide usar una entidad de seguridad para autenticar y autorizar con el identificador de Entra de Microsoft para acceder a un recurso de Azure, consulte Autorizar el acceso con el identificador de Entra de Microsoft para asegurarse de que a la entidad de seguridad se le ha concedido el permiso suficiente para acceder al recurso de Azure.

Propiedades configurables de conexión de spring-cloud-azure-starter-integration-storage-queue:

Propiedad	Tipo	Descripción
spring.cloud.azure.storage.queue.enabled	booleano	Si una cola de Azure Storage está habilitada.
spring.cloud.azure.storage.queue.connection-string	Cuerda	Valor de la cadena de conexión del espacio de nombres de la cola de almacenamiento.
spring.cloud.azure.storage.queue.accountName	Cuerda	Nombre de la cuenta de cola de Storage.
spring.cloud.azure.storage.queue.accountKey	Cuerda	Clave de cuenta de Storage Queue.
spring.cloud.azure.storage.queue.endpoint	Cuerda	Punto de conexión de servicio de Storage Queue.
spring.cloud.azure.storage.queue.sasToken	Cuerda	Credencial del token de Sas
spring.cloud.azure.storage.queue.serviceVersion	QueueServiceVersion	QueueServiceVersion que se usa al realizar solicitudes de API.
spring.cloud.azure.storage.queue.messageEncoding	Cuerda	Codificación de mensajes de cola.

Uso básico

Envío de mensajes a la cola de Azure Storage

1. Rellene las opciones de configuración de credenciales.

   • Para las credenciales como cadena de conexión, configure las siguientes propiedades en el archivo application.yml:

     spring:
       cloud:
         azure:
           storage:
             queue:
               connection-string: ${AZURE_STORAGE_QUEUE_CONNECTION_STRING}
     

     Nota

     Microsoft recomienda usar el flujo de autenticación más seguro disponible. El flujo de autenticación descrito en este procedimiento, como para bases de datos, memorias caché, mensajería o servicios de inteligencia artificial, requiere un grado de confianza muy alto en la aplicación y conlleva riesgos que no están presentes en otros flujos. Use este flujo solo cuando las opciones más seguras, como las identidades administradas para conexiones sin contraseña o sin claves, no sean viables. En el caso de las operaciones de máquina local, prefiera identidades de usuario para conexiones sin contraseña o sin claves.

   • Para las credenciales como identidades administradas, configure las siguientes propiedades en el archivo 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}
     

Nota

Los valores permitidos para tenant-id son: common, organizations, consumerso el identificador de inquilino. Para obtener más información sobre estos valores, consulte la sección Uso del punto de conexión incorrecto (cuentas personales y de organización) de Error AADSTS50020: la cuenta de usuario del proveedor de identidades no existe en el inquilino. Para obtener información sobre cómo convertir la aplicación de un solo inquilino, consulte Convertir aplicación de un solo inquilino en multiinquilino en microsoft Entra ID.

• Para las credenciales como principal de servicio, configure las siguientes propiedades en su archivo 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}
  

Nota

Los valores permitidos para tenant-id son: common, organizations, consumerso el identificador de inquilino. Para obtener más información sobre estos valores, consulte la sección Uso del punto de conexión incorrecto (cuentas personales y de organización) de Error AADSTS50020: la cuenta de usuario del proveedor de identidades no existe en el inquilino. Para obtener información sobre cómo convertir la aplicación de un solo inquilino, consulte Convertir aplicación de un solo inquilino en multiinquilino en microsoft Entra ID.

1. Cree DefaultMessageHandler con el bean StorageQueueTemplate para enviar mensajes a Storage Queue.

   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.error("There was an error sending the message.", ex);
               }
           });
           return handler;
       }
   }
   

2. Cree una vinculación de puerta de enlace de mensajes con el controlador de mensajes anterior a través de un canal de mensajes.

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

3. Enviar mensajes mediante la puerta de enlace.

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

Recibir mensajes desde Azure Storage Queue

1. Rellene las opciones de configuración de credenciales.

2. Cree un bean del canal de mensajes como canal de entrada.

   class Demo {
       private static final String INPUT_CHANNEL = "input";
   
       @Bean
       public MessageChannel input() {
           return new DirectChannel();
       }
   }
   

3. Cree StorageQueueMessageSource con el bean StorageQueueTemplate para recibir mensajes en Storage Queue.

   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);
       }
   }
   

4. Cree un enlace de receptor de mensajes con StorageQueueMessageSource creado en el último paso a través del canal de mensajes que creamos antes.

   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();
       }
   }
   

Muestras

Para obtener más información, consulte el azure-spring-boot-samples repositorio en GitHub.