如何处理带有 DUPS_OK_ACKNOWLEDGE 异常的 jms 消息使用者

Question

我正在使用来自 tibco 队列的消息，session transacted true 的吞吐量为 13 msgs/sec，session transacted as false 和 DUPS_OK_ACKNOWLEDGE ack 模式的吞吐量为 160 msgs/sec，这是有希望的，但是当应用程序中抛出异常，然后消息丢失。

任何人都可以建议我如何处理这种情况，我需要高吞吐量，同时不会在异常时丢失消息。

    .from(Jms.messageDrivenChannelAdapter(tibcoConnectionFactory)
            .destination(sourceQueue)
            .configureListenerContainer(spec -> {
                spec.sessionTransacted(false);
                spec.sessionAcknowledgeMode(Session.DUPS_OK_ACKNOWLEDGE);
            }))
            .transform(orderTransformer, "transform", e -> e.advice(idempotentReceiverInterceptor())
            .handle(orderService, "save")
            .get();

Answer 1

.from(Jms.messageDrivenChannelAdapter(tibcoConnectionFactory)

将使用默认容器，如果不进行事务处理，该容器可能会丢失消息，因为侦听器是在 consumer.receive() 之外调用的。

使用

.from(Jms.messageDrivenChannelAdapter(tibcoConnectionFactory, 
    SimpleMessageListenerContainer.class)

改为。

编辑

错误信息应该是一个线索...

2018-05-18 11:38:43.657 WARN 48531 --- [Session Task-1] o.s.j.l.SimpleMessageListenerContainer：JMS 消息监听器的执行失败，并且没有设置 ErrorHandler。

所以添加一个错误处理程序...

@SpringBootApplication
public class So50413144Application {

    public static void main(String[] args) {
        SpringApplication.run(So50413144Application.class, args);
    }

    @Bean
    public ApplicationRunner runner(JmsTemplate template) {
        return args -> {
            for (int i = 0; i < 10; i++) {
                template.convertAndSend("foo", "foo" + i);
            }
        };
    }

    @Bean
    public IntegrationFlow flow(ConnectionFactory connectionFactory) {
        return IntegrationFlows
                .from(Jms.messageDrivenChannelAdapter(connectionFactory, SimpleMessageListenerContainer.class)
                        .destination("foo")
                        .configureListenerContainer(c -> {
                            c.sessionAcknowledgeMode(Session.DUPS_OK_ACKNOWLEDGE);
                            c.errorHandler(t -> {
                                if (t instanceof RuntimeException) {
                                    throw (RuntimeException) t;
                                }
                            });
                        }))
                .handle((p, h) -> {
                    System.out.println(p);
                    if (p.equals("foo5")) {
                        throw new RuntimeException("fail");
                    }
                    try {
                        Thread.sleep(200);
                    }
                    catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                    return null;
                })
                .get();
    }

}

...和foo5 一遍又一遍地传递。