Guía de solución de problemas de Spring JMS

En este artículo se describe cómo solucionar problemas conocidos y errores comunes al usar Spring JMS. El artículo también responde a algunas preguntas más frecuentes sobre spring-cloud-azure-starter-servicebus-jms.

Problemas de conectividad

MessageProducer se cerró debido a un error irrecuperable

Descripción del problema

Al usar JmsTemplate para enviar mensajes, JmsTemplate deja de estar disponible durante un intervalo de inactividad comprendido entre 10 y 15 minutos. Enviar mensajes en ese intervalo puede obtener las excepciones que se muestran en la siguiente salida de ejemplo:

2022-11-06 11:12:05.762  INFO 25944 --- [   scheduling-1] c.e.demo.ServiceBusJMSMessageProducer    : Sending message: 2022-11-06T11:12:05.762072 message 1
2022-11-06 11:12:05.772 ERROR 25944 --- [   scheduling-1] o.s.s.s.TaskUtils$LoggingErrorHandler    : Unexpected error occurred in scheduled task

org.springframework.jms.IllegalStateException: The MessageProducer was closed due to an unrecoverable error.; nested exception is javax.jms.IllegalStateException: The MessageProducer was closed due to an unrecoverable error.
    at org.springframework.jms.support.JmsUtils.convertJmsAccessException(JmsUtils.java:274) ~[spring-jms-5.3.23.jar:5.3.23]
  ...
Caused by: org.apache.qpid.jms.provider.ProviderException: The link 'G0:36906660:qpid-jms:sender:azure:5caf3ef4-9602-413c-964d-cf1292d6e1f5:1:1:1:t4' is force detached. Code: publisher(link376). Details: AmqpMessagePublisher.IdleTimerExpired: Idle timeout: 00:10:00. [condition = amqp:link:detach-forced]
    at org.apache.qpid.jms.provider.amqp.AmqpSupport.convertToNonFatalException(AmqpSupport.java:181) ~[qpid-jms-client-0.53.0.jar:na]
  ...

Análisis de causa

Las excepciones se producen en Azure Service Bus cuando la conexión y el enlace de AMQP están activos, pero no se realiza ninguna llamada —por ejemplo, de envío o de recepción— a través del enlace durante 10 minutos. En este caso, se cierra el vínculo. Y cuando se han cerrado todos los vínculos de la conexión porque no había ninguna actividad (inactiva) y no se ha creado un vínculo nuevo en 5 minutos, se cierra la conexión.

Para el iniciador de JMS de Service Bus, se utiliza de forma predeterminada la CachingConnectionFactory, que almacena en caché la sesión, el productor y el consumidor. Cuando el JmsProducer permanece inactivo más de 10 minutos pero menos de 15, se ha cerrado el enlace que estaba ocupando el productor en caché. Los mensajes no se pueden enviar durante este intervalo. A continuación, después de otros 5 minutos inactivos, se cierra toda la conexión. Por lo tanto, cualquier operación de envío después del intervalo de inactividad de 15 minutos hace que el CachingConnectionFactory cree una nueva conexión para enviar. La operación de envío estará disponible después de 15 minutos.

Solución alternativa

Actualmente, el iniciador proporciona una solución alternativa para el problema de desasociación de vínculos aplicando JmsPoolConnectionFactory, que agrupa Connection, Session y MessageProducer, y administra el ciclo de vida de las instancias agrupadas. Esta solución alternativa puede garantizar que un productor se expulse después de no estar disponible y, por tanto, todas las operaciones de envío se realizan en productores activos.

Para usar esta solución alternativa, agregue la siguiente configuración:

spring:
  jms:
    servicebus:
      pool:
        enabled: true
        max-connections: ${your-expected-max-connection-value}

Uso de spring.jms.servicebus.idle-timeout

Las propiedades de tiempo de espera de inactividad configuran el tiempo de espera de inactividad de una conexión AMQP. La especificación AMQP proporciona la siguiente descripción:

Las conexiones están sujetas a un umbral de tiempo de inactividad. Un par local desencadena el tiempo de espera cuando no se reciben fotogramas después de superar un valor umbral. El tiempo de espera de inactividad se mide en milisegundos y comienza desde la hora en que se recibe el último fotograma. Si se supera el umbral, un elemento del mismo nivel DEBE intentar cerrar correctamente la conexión mediante un marco de cierre con un error que explica por qué. Si el par remoto no responde de forma ordenada dentro del plazo establecido, el par PUEDE cerrar el socket TCP.

En el caso de un cliente JMS, al configurar esta propiedad, controlará en el servidor cuánto tiempo espera que el servidor envíe un marco vacío para mantener activa una conexión cuando no se entrega ningún mensaje. Esta propiedad controla el comportamiento del par remoto, y cada par puede tener su propio valor independiente.

Problemas de JmsTemplate

Mensajes programados

Azure Service Bus admite el procesamiento de mensajes retrasado. Para obtener más información, consulte la sección Mensajes programados de Secuenciación de mensajes y marcas de tiempo. Para JMS, para programar un mensaje, establezca la propiedad ScheduledEnqueueTimeUtc mediante el encabezado de anotación de mensaje x-opt-scheduled-enqueue-time.

Problemas de JmsListener

Se envían demasiadas solicitudes a Service Bus aunque no haya ningún mensaje en el servidor.

Descripción del problema

Al usar la API de @JmsListener, en algunos casos puede ver en el portal de Azure valores activos para las solicitudes entrantes enviadas a sus colas o temas, aunque no haya mensajes en el servidor para recibir.

Análisis de causa

@JmsListener es un agente de escucha de sondeo, diseñado para repetir sondeos.

El agente de escucha se encuentra en un bucle de sondeo en curso. Cada iteración llama al método MessageConsumer.receive() de JMS para sondear al consumidor local en busca de mensajes para consumir. De forma predeterminada, para cada operación de sondeo, el consumidor local envía solicitudes de incorporación de cambios al agente de mensajes para solicitar mensajes y, a continuación, bloquea durante un período de tiempo determinado. El proceso específico de sondeo viene determinado por varias propiedades, incluidas receiveTimeout, prefetchSize y receiveLocalOnly o receiveNoWaitLocalOnly. El método receiveNoWaitLocalOnly solo se usa cuando se establece receiveTimeout en un valor negativo.

Cuando se produzca este problema en la aplicación, compruebe las siguientes opciones de configuración:

  • Determine si la directiva de captura previa es 0, que también es la opción predeterminada. Una captura previa de 0 significa un consumidor de extracción que envía solicitudes de extracción al Service Bus en cada sondeo.

  • Si ha configurado una captura previa distinta de cero, determine si las propiedades receiveLocalOnly o receiveNoWaitLocalOnly se deben configurar como false, que es la opción predeterminada. Un valor de false aquí sigue dando como resultado el envío de pull requests al servidor porque no se limita a sondear al consumidor local.

  • La configuración de receiveTimeout determina cuánto tiempo bloquea cada solicitud de incorporación de cambios, por lo que puede afectar a la frecuencia de las solicitudes de incorporación de cambios que se envían al servidor. El valor predeterminado es 1 segundo.

Para obtener un análisis completo, consulte la explicación del problema de GitHub .

Soluciones

En las secciones siguientes se describen dos soluciones para tratar este problema.

Solución 1. Cambio para insertar solo consumidor y comprobación local

Al cambiar el modo a push, el consumidor se convierte en un consumidor de notificación asíncrona que no extrae mensajes del bróker, sino que mantiene una cantidad objetivo de crédito de enlace. La cantidad viene determinada por una propiedad de precarga. A medida que Service Bus (remitente) inserta mensajes, el crédito de vínculo del remitente disminuye y, cuando el crédito de vínculo del remitente está por debajo de un umbral, el cliente (receptor) envía una solicitud al servidor para aumentar el crédito de vínculo del remitente a la cantidad de destino deseada.

Para realizar esta solución, agregue la siguiente configuración:

Primero, configure prefetch con un valor distinto de cero, lo que configura el consumidor sin extracción. En la tabla siguiente se muestran varias propiedades de captura previa, cada una de las cuales controla diferentes entidades de Service Bus. Establezca las propiedades que se aplican a su caso.

Propiedad Descripción
spring.jms.servicebus.prefetch.all El valor predeterminado de la opción de prefetch en este espacio de nombres de Service Bus
spring.jms.servicebus.prefetch.queue-prefetch Número de captura previa de la cola.
spring.jms.servicebus.prefetch.queue-browser-prefetch Número de captura previa del explorador de colas.
spring.jms.servicebus.prefetch.topic-prefetch Número de captura previa del tema.
spring.jms.servicebus.prefetch.durable-topic-prefetch Número de captura previa del tema duradero.

En segundo lugar, configure el non-local-check agregando una clase de configuración para el personalizador de fábrica, como se muestra en el ejemplo siguiente:

@Configuration(proxyBeanMethods = false)
public class CustomJmsConfiguration {

    @Bean
    ServiceBusJmsConnectionFactoryCustomizer customizer() {
        return factory -> {
            factory.setReceiveLocalOnly(true);
            // Configure the below ReceiveNoWaitLocalOnly instead if you have specified the property 
            // spring.jms.listener.receive-timeout with negative value. Otherwise, configure the above `ReceiveLocalOnly`.
            //factory.setReceiveNoWaitLocalOnly(true);
        };
    }
}

El valor de captura previa puede afectar a la rapidez de envío de mensajes al búfer local del consumidor. Debe ajustar el valor según el rendimiento consumido y los volúmenes de mensajes. Un valor adecuado puede acelerar el proceso de consumo, mientras que un valor demasiado grande puede hacer que los mensajes almacenados en búfer local se vuelvan obsoletos y se envíen de nuevo. Para volúmenes de mensajes bajos, donde cada mensaje tarda mucho tiempo en procesarse, establezca la captura previa en 1. Este valor garantiza que un consumidor solo está procesando un mensaje cada vez.

Solución 2. Aumento del tiempo de espera de recepción para reducir la frecuencia de extracción

La propiedad de tiempo de espera de recepción determina la estrategia durante cuánto tiempo bloquea el consumidor para esperar un resultado de extracción. Por lo tanto, al extender el tiempo de espera, puede reducir la frecuencia de extracción y, a continuación, el número de solicitudes de extracción al elegir el modo de extracción. En casos extremos, puede establecer la estrategia para esperar indefinidamente hasta que llegue un mensaje, lo que significa que el consumidor solo extrae después de consumir un mensaje. En este caso, cuando no haya ningún mensaje en el servidor, se bloqueará para esperar.

Para realizar esta solución, configure la propiedad spring.jms.listener.receive-timeout. Esta propiedad es de tipo java.time.Duration y tiene un valor predeterminado de 1 segundo. En la lista siguiente se explica el efecto de varios valores:

  • Establecer el tiempo de espera de recepción en 0 significa que la operación de extracción queda bloqueada indefinidamente hasta que se entrega un mensaje.
  • Establecer el tiempo de espera de recepción en un valor positivo significa que la operación de extracción queda bloqueada hasta que se agota el tiempo de espera.
  • Establecer el tiempo de espera de recepción en un valor negativo significa que la operación de extracción es una recepción sin espera; es decir, devuelve un mensaje inmediatamente o null si no hay mensajes disponibles.

Nota

Un valor de tiempo de espera elevado puede aportar algunos efectos secundarios. Por ejemplo, un valor elevado de tiempo de espera también prolongará el tiempo durante el que el hilo principal permanece en estado de bloqueo. Este estado significa que el contenedor tendrá menos capacidad de respuesta a las llamadas stop() y solo puede detenerse entre receive() llamadas.

Además, el contenedor solo puede enviar solicitudes después de que se haya superado el intervalo de receive-timeout. Si el intervalo es superior a 10 minutos, Service Bus cerrará el vínculo e impedirá que el agente de escucha envíe o reciba. Para más información, consulte la sección Link is closed de AMQP errors in Azure Service Bus. De forma predeterminada, el escucha usa una CachingConnectionFactory.

Si necesita un tiempo de espera de recepción elevado, asegúrese de usar el JmsPoolConnectionFactory.

Para obtener más información sobre el problema de cierre del vínculo y cómo usar JmsPoolConnectionFactory, vea la Se cerró MessageProducer debido a un error irrecuperable sección.

Problema de captura previa

Descripción del problema

Una directiva de captura previa no adecuada puede provocar los siguientes problemas:

  • Los mismos mensajes se consumen repetidamente.
  • Los mensajes se colocan en la cola de mensajes fallidos después de MaxDeliveryCountExceeded, incluso cuando los mensajes se procesan sin error o excepción.

Análisis de causa

Este problema suele ocurrir cuando el valor de prefetch es mayor que la capacidad de consumo real, lo que provoca que se precarguen demasiados mensajes en el búfer local, a la espera de ser consumidos. Sin embargo, los mensajes con captura previa se consideran enviados en modo peek-lock por parte de Service Bus. Cada mensaje enviado tiene un max-delivery-count atributo y lock-duration . En el modo de recepción peek-lock, los mensajes capturados en el búfer de captura previa se adquieren en el búfer en un estado bloqueado y se selecciona el reloj de tiempo de espera para el bloqueo. Si el búfer de captación previa es grande y el procesamiento tarda tanto que los bloqueos de los mensajes expiran mientras permanecen en ese búfer, el mensaje se considera abandonado y vuelve a estar disponible para recuperarse de la cola.

Este problema puede hacer que el mensaje se capture en el búfer de captura previa y se coloque al final. Si el búfer de captura previa no se procesa antes de la expiración del mensaje, los mensajes se capturan previamente repetidamente, pero nunca se entregan de forma efectiva en un estado utilizable (bloqueado válidamente). A continuación, cuando esas copias obsoletas se desencolan, la aplicación vuelve a consumir el mismo mensaje repetidamente y no puede completar su procesamiento. En otro caso, todos los mensajes repetidos expiran en el búfer antes de consumirse. En este caso, el mensaje en Service Bus acabará moviéndose a la cola de mensajes fallidos cuando se supere el número máximo de entregas.

Para obtener más información, consulte la sección ¿Por qué la captura previa no es la opción predeterminada? de Captura previa de mensajes de Azure Service Bus.

Solución

Tenga cuidado con la configuración del prefetch para que se ajuste a la capacidad de procesamiento del consumidor. Debe equilibrar el número máximo de captura previa y la duración del bloqueo configurado en la cola o suscripción de forma que el tiempo de espera del bloqueo supere al menos el tiempo de procesamiento de mensajes acumulativo para el tamaño máximo del búfer de captura previa, además de un mensaje. Al mismo tiempo, el tiempo de expiración del bloqueo no debe ser tan largo como para que los mensajes puedan superar su tiempo de vida máximo cuando se descartan accidentalmente y, por tanto, sea necesario esperar a que su bloqueo expire antes de que se vuelvan a entregar.

Para configurar el atributo de captura previa, que tiene un valor predeterminado de cero, use una de las siguientes propiedades:

Propiedad Descripción
spring.jms.servicebus.prefetch.all Valor de reserva para la opción de captura previa en este espacio de nombres de Service Bus.
spring.jms.servicebus.prefetch.queue-prefetch Número de captura previa de la cola.
spring.jms.servicebus.prefetch.queue-browser-prefetch Número de captura previa del explorador de colas.
spring.jms.servicebus.prefetch.topic-prefetch Número de captura previa del tema.
spring.jms.servicebus.prefetch.durable-topic-prefetch Número de captura previa del tema duradero.

¿Cómo se realiza una disposición de AMQP en Service Bus?

JMS admite cinco tipos de disposición de AMQP al confirmar mensajes ante el broker de mensajería. Los valores admitidos son ACCEPTED, REJECTED, RELEASED, MODIFIED_FAILEDy MODIFIED_FAILED_UNDELIVERABLE. Para más información, consulte la sección asignación de operaciones de AmQP y Service Bus de Uso de Java Message Service 1.1 con azure Service Bus estándar y AMQP 1.0.

Por lo tanto, para completar manualmente, abandonar, enviar, aplazar o liberar un mensaje mediante JmsListener, siga estos pasos:

  1. Deshabilite las transacciones de sesión y use el modo de confirmación de CLIENTE.

    Para realizar esta tarea, o bien declare su propio bean JmsListenerContainerFactory y, a continuación, establezca las propiedades, o bien realice el posprocesamiento de JmsListenerContainerFactory definido en el starter. En el siguiente ejemplo se utiliza el método de declarar otro bean:

    @Configuration(proxyBeanMethods = false)
public class CustomJmsConfiguration {

    @Bean
    public JmsListenerContainerFactory<?> customQueueJmsListenerContainerFactory(
            DefaultJmsListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory jmsListenerContainerFactory = new DefaultJmsListenerContainerFactory();
        configurer.configure(jmsListenerContainerFactory, connectionFactory);
        jmsListenerContainerFactory.setPubSubDomain(Boolean.FALSE);
        jmsListenerContainerFactory.setSessionTransacted(Boolean.FALSE);
        jmsListenerContainerFactory.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
        return jmsListenerContainerFactory;
    }
}

  2. En el controlador de mensajes, complete o abandone mensajes explícitamente.

    @JmsListener(destination = "QUEUE_NAME", containerFactory = "customQueueJmsListenerContainerFactory")
public void receiveMessage(JmsTextMessage message) throws Exception {
    String event = message.getBody(String.class);
    try {
        logger.info("Received event: {}", event);
        logger.info("Received message: {}", message);
        // by default complete the message
        message.acknowledge();
    } catch (Exception e) {
        logger.error("Exception while processing re-source event: " + event, e);
        JmsAcknowledgeCallback acknowledgeCallback = message.getAcknowledgeCallback();
        // explicitly abandon the message
        acknowledgeCallback.setAckType(MODIFIED_FAILED);
        message.setAcknowledgeCallback(acknowledgeCallback);
        message.acknowledge();
        throw e;
    }
}

Problemas de configuración

Deshabilitar la configuración automática de JMS de Service Bus

Descripción del problema

Algunos usuarios importan algún Spring Cloud Azure Starter para la autoconfiguración de un servicio de Azure que no sea Service Bus JMS. También usan el framework Spring JMS sin necesidad de usar Service Bus JMS. A continuación, cuando la aplicación intenta iniciarse, se producen las siguientes excepciones:

Caused by: java.lang.IllegalArgumentException: 'spring.jms.servicebus.connection-string' should be provided
    at com.azure.spring.cloud.autoconfigure.jms.properties.AzureServiceBusJmsProperties.afterPropertiesSet(AzureServiceBusJmsProperties.java:210)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.invokeInitMethods(AbstractAutowireCapableBeanFactory.java:1863)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1800)
    ... 98 more

Análisis de causa

Este problema se produce porque todas las clases de autoconfiguración de Spring Cloud Azure se encuentran en el mismo módulo, por lo que cualquier Spring Cloud Azure Starter importa toda esa autoconfiguración, que también incluye Service Bus JMS. A continuación, cuando la aplicación usa la API de Spring JMS, cumple la condición de configuración automática de JMS de Service Bus y la desencadena. A continuación, para los usuarios que no piensan usar spring-cloud-azure-starter-servicebus-jms, no se cumplen las condiciones de propiedad porque no hay ninguna razón para configurar Service Bus para JMS. Esta situación provoca que se lancen excepciones.

Solución

Spring Cloud Azure for Service Bus JMS proporciona una propiedad para activar o desactivar su configuración automática. Puede optar por deshabilitar esta funcionalidad según sea necesario mediante el siguiente valor de propiedad:

spring.jms.servicebus.enabled=false

Configuración de atributos de mensaje

¿Cómo establecer el tipo de contenido de los mensajes salientes?

Para configurar el tipo de contenido, personalice el convertidor de mensajes para modificar el atributo de tipo de contenido al convertir mensajes. El código siguiente toma mensajes de bytes como ejemplo.

En primer lugar, personalice el convertidor de mensajes que se usará en el JmsTemplate, como se muestra en el ejemplo siguiente:

public class CustomMappingJackson2MessageConverter extends MappingJackson2MessageConverter {

    public static final String CONTENT_TYPE = "application/json";

    public CustomMappingJackson2MessageConverter() {
        this.setTargetType(MessageType.BYTES);
    }

    @Override
    protected BytesMessage mapToBytesMessage(Object object, Session session, ObjectWriter objectWriter)
            throws JMSException, IOException {
        final BytesMessage message = super.mapToBytesMessage(object, session, objectWriter);
        JmsBytesMessage msg = (JmsBytesMessage) message;
        AmqpJmsMessageFacade facade = (AmqpJmsMessageFacade) msg.getFacade();
        facade.setContentType(Symbol.valueOf(CONTENT_TYPE));
        return msg;
    }
}

A continuación, declare el bean personalizado del convertidor de mensajes, como se muestra en este ejemplo:

@Configuration(proxyBeanMethods = false)
public class CustomJmsConfiguration {

    @Bean
    public MessageConverter messageConverter() {
        return new CustomMappingJackson2MessageConverter();
    }
}

¿Cómo establecer el nombre de propiedad de identificador de tipo para MappingJackson2MessageConverter?

El atributo type-id-property-name permite al MappingJackson2MessageConverter determinar qué clase usar para deserializar la carga del mensaje. Al serializar cada objeto Java como carga útil de un mensaje de Spring, el convertidor almacena el tipo de la carga útil en una propiedad del mensaje cuyo nombre queda registrado mediante type-id-property-name. A continuación, al deserializar el mensaje, el convertidor lee el identificador de tipo del mensaje y realiza la deserialización.

Para establecer el type-id-property-name, declare su propio bean MappingJackson2MessageConverter y configure esa propiedad, como se muestra en el ejemplo siguiente:

@Configuration(proxyBeanMethods = false)
public class CustomJmsConfiguration {

    @Bean
    public MessageConverter jacksonJmsMessageConverter()
    {
        MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
        converter.setTypeIdPropertyName("your-custom-type-id-property-name");
        return converter;
    }
}

Detección de duplicados

Azure Service Bus admite detección de duplicados, que aplica la propiedad MessageId para identificar de forma única los mensajes y descartar los duplicados enviados a Service Bus.

Sin embargo, para la API de JMS, no debe asignar el identificador de mensaje de JMS, que se considera ilegal en las especificaciones de JMS. Por lo tanto, esta característica no se admite actualmente para Spring Cloud Azure Service Bus JMS Starter.

Para cualquier actualización futura sobre esta función, consulte la issue de GitHub.

Habilitar el registro de transporte de AMQP

Para obtener más información, consulte la sección habilitar el registro de transporte de AMQP de Solución de problemas de Service Bus.

Obtener ayuda adicional

Para obtener más información sobre las formas de ponerse en contacto con el soporte técnico, consulte soporte técnico en la raíz del repositorio.

Recursos para el inicio de JMS de Spring Cloud Azure Service Bus

Presentación de problemas de GitHub

Al presentar problemas de GitHub, se solicitan los detalles siguientes:

  • Configuración de Service Bus / Entorno del espacio de nombres
    • ¿Cuál es el nivel de servicio del espacio de nombres (Estándar o Premium)?
    • ¿Qué tipo de entidad de mensajería se está utilizando (cola o tema)? y su configuración.
    • ¿Cuál es el tamaño medio de cada mensaje?
  • ¿Cuál es el patrón de tráfico? (es decir, el número de mensajes por minuto y si el cliente siempre está ocupado o tiene períodos de tráfico lentos).
  • Código y pasos de reproducción
    • Esto es importante, ya que a menudo no podemos reproducir el problema en nuestro entorno.
  • Registros
  • Se actualizó por última vez el