我的集成流程定义如下:
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 成功完成之前不确认任何消息吗？
最佳答案
您可以使用手动确认模式并随后通过 header 确认:
关于java - 在 IntegrationFlow 服务激活器方法成功返回之前不确认 RabbitMQ 消息？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/47455784/