如何使用 Spring JmsListener 手动确认来自 ActiveMQ 的消息

Question

我正在使用带有 Spring 的 JmsListener 的 ActiveMQ（带有 JMS）。我可以使用来自 ActiveMQ 队列的消息，但它使用的是AUTO_ACKNOWLEDGE。如何设置 CLIENT_ACKNOWLEDGE 以便只有在确认后才能使用另一条消息。

@Bean
public ActiveMQConnectionFactory connectionFactory() {
    ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory();
    activeMQConnectionFactory.setTransactedIndividualAck(true);
    activeMQConnectionFactory.setUserName(mqUserName);
    activeMQConnectionFactory.setPassword(mqPassword);
    activeMQConnectionFactory.setBrokerURL(mqUrl);
    return activeMQConnectionFactory;
}

@Bean
public JmsListenerContainerFactory myFactory(ConnectionFactory connectionFactory, DefaultJmsListenerContainerFactoryConfigurer configurer) {

    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    factory.setErrorHandler(t -> {
        logger.info("An error has occurred in the transaction");
        logger.error(t.getCause().getMessage());
    });

    configurer.configure(factory, connectionFactory);
    factory.setConcurrency("4");

    // You could still override some of Boot's default if necessary.
    return factory;
}

@Bean
public MessageConverter jacksonJmsMessageConverter() {
    MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
    converter.setTargetType(MessageType.TEXT);
    converter.setTypeIdPropertyName("_type");
    return converter;
}

@JmsListener(destination = "QUEUE_1", containerFactory = "myFactory", concurrency = "2")
public void receiveImgGenerationMessage(String transaction) {
    logger.info("message received in queue " + transaction);
    //I will call other api to process the message and do some operation 
    //after the message is processed 
    //I have to Acknowledge the message is processed
    //so that i can consume the other message for process.
}

// jmsTemplate bean
public void sendmessage() {
    for (int i =0 ; i < 10 ; < i++) { 
        jmsTemplate.convertAndSend("QUEUE_1", i);
    }
}

Answer 1

您应该在您的设备上使用setSessionAcknowledgeMode 方法 org.springframework.jms.config.DefaultJmsListenerContainerFactory 实例设置CLIENT_ACKNOWLEDGE 模式，例如：

@Bean
public JmsListenerContainerFactory myFactory(ConnectionFactory connectionFactory, DefaultJmsListenerContainerFactoryConfigurer configurer) {

    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    factory.setErrorHandler(t -> {
        logger.info("An error has occurred in the transaction");
        logger.error(t.getCause().getMessage());
    });

    factory.setSessionAcknowledgeMode(javax.jms.Session.CLIENT_ACKNOWLEDGE);

    configurer.configure(factory, connectionFactory);
    factory.setConcurrency("4");

    // You could still override some of Boot's default if necessary.
    return factory;
}

这在Spring JMS JavaDoc中讨论：

侦听器容器提供以下消息确认选项：

• “sessionAcknowledgeMode”设置为“AUTO_ACKNOWLEDGE”（默认）：此模式依赖于容器：对于DefaultMessageListenerContainer，这意味着在侦听器执行之前自动确认消息，并且没有重新传递在其他侦听器执行中断的情况下也不会重新传递。对于SimpleMessageListenerContainer，这意味着在侦听器执行之后 自动确认消息，在抛出用户异常的情况下不会重新发送，但在侦听器执行期间JVM 死机的情况下可能会重新发送。为了始终如一地安排任何容器变体的重新交付，请考虑“CLIENT_ACKNOWLEDGE”模式或 - 最好 - 将“sessionTransacted”设置为“true”。

• “sessionAcknowledgeMode”设置为“DUPS_OK_ACKNOWLEDGE”：在 (DefaultMessageListenerContainer) 期间或 (SimpleMessageListenerContainer) 侦听器执行后不久的 Lazy 消息确认；如果抛出用户异常，则不会重新交付，但如果 JVM 在侦听器执行期间死亡，则可能会重新交付。为了始终如一地安排任何容器变体的重新交付，请考虑“CLIENT_ACKNOWLEDGE”模式或 - 最好 - 将“sessionTransacted”设置为“true”。

• “sessionAcknowledgeMode”设置为“CLIENT_ACKNOWLEDGE”：在侦听器成功执行之后自动消息确认；在引发用户异常以及其他侦听器执行中断（例如 JVM 死机）的情况下，尽最大努力重新交付。

• “sessionTransacted”设置为“true”：监听器执行成功后的事务确认； 保证重新交付，以防引发用户异常以及其他侦听器执行中断（例如 JVM 死机）。

您也可以在 Spring Boot application.properties 中使用它：

spring.jms.listener.acknowledge-mode=CLIENT