如何使用 Spring JmsListener 手动确认消息

Question

我将 JMS 与 Spring 的 JmsListener 一起使用。我能够使用 JMS 队列中的消息，但它正在使用AUTO_ACKNOWLEDGE. 我如何设置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());
    });

    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自动消息确认，在发生异常的情况下不会重新传递，并且在其他侦听器执行中断的情况下也不会重新传递。对于，它意味着在侦听器执行后自动消息确认，在抛出用户异常的情况下不会重新传递，但在侦听器执行期间 JVM 死亡的情况下可能会重新传递。为了一致地安排任何容器变体的重新交付，请考虑“CLIENT_ACKNOWLEDGE”模式，或者最好将“sessionTransacted”设置为“true”。SimpleMessageListenerContainer

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

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

• “sessionTransacted”设置为“true”：侦听器成功执行后的事务确认；在抛出用户异常以及其他侦听器执行中断（例如 JVM 死亡）的情况下保证重新传递。

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

spring.jms.listener.acknowledge-mode=CLIENT