在 IntegrationFlow 服务激活器方法成功返回之前不确认 RabbitMQ 消息？

Question

我有一个这样定义的集成流程：

IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, "queueName")
                    .id("id")
                    .autoStartup(autoStartup)
                    .concurrentConsumers(2)
                    .maxConcurrentConsumers(3)
                    .messageConverter(messageConverter()))
                    .aggregate(a -> ...)
                    .handle(serviceActivatorBean, "myMethod", e -> e.advice(requestHandlerRetryAdviceForIntegrationFlow()))
                    .get();

serviceActivatorBean 的定义如下：

@Component
@Transactional
public class ServiceActivator {

    @ServiceActivator    
    public void myMethod(Collection<MyEvent> events) {
        ....
    }    
}

而requestHandlerRetryAdviceForIntegrationFlow() 是这样定义的：

  public static RequestHandlerRetryAdvice requestHandlerRetryAdviceForIntegrationFlow() {
    RequestHandlerRetryAdvice advice = new RequestHandlerRetryAdvice();
    RetryTemplate retryTemplate = new RetryTemplate();
    SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
    retryPolicy.setMaxAttempts(MAX_VALUE);
    retryTemplate.setRetryPolicy(retryPolicy);
    retryTemplate.setListeners(new RetryListenerSupport[]{new RetryListenerSupport() {
        @Override
        public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) {
            log.error("Caught exception {} (retry count {}), will retry again!", throwable.getClass().getSimpleName(),
                    context.getRetryCount(), throwable);
        }
    }});
    advice.setRetryTemplate(retryTemplate);
    ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
    backOffPolicy.setMaxInterval(5000L);
    backOffPolicy.setInitialInterval(200L);
    backOffPolicy.setMultiplier(2);
    retryTemplate.setBackOffPolicy(backOffPolicy);
    return advice;
}

我们面临的问题是当服务激活器中的events 集合包含2 个或更多事件并且由于某种原因myMethod 的处理失败并且服务器崩溃时。似乎发生的事情是IntegrationFlow 一次从 RabbitMQ 消费和确认一条消息，因此如果在处理myMethod 期间服务器崩溃，除了最后一个事件之外的所有事件都会丢失。这对我们来说既不好也不安全。我们可以做些什么来配置IntegrationFlow 在服务激活器中的myMethod 成功完成之前不确认任何消息？

Answer 1

您可以使用 Acknowledge Mode MANUAL 并随后通过标题确认：

https://docs.spring.io/spring-integration/docs/4.3.12.RELEASE/reference/html/amqp.html#amqp-inbound-channel-adapter