在用于AMQP的Spring集成中使用ImmediateRequeueMessageRecoverer？

Question

我们已经注意到，当错误消息被接收到Spring集成endpoint（从RabbitMQ）时，它们不会被重试。如果我们的业务代码（即接收消息的“服务方法”）出现了问题，导致它抛出异常，那么重试就会发生。

这是我们的配置：

var myService = ...
IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, queueName)
                .id(integrationFlowId)
                .autoStartup(autoStartup)
                .configureContainer(c -> c.acknowledgeMode(MANUAL)
                        .prefetchCount(10)
                        .concurrentConsumers(1)
                        .maxConcurrentConsumers(3))
                .messageConverter(messageConverter))
                .aggregate(...)
                .handle(myService, "myMethod", e -> e.advice(myAdvice()))
                .get();

myadvice方法的实现方式如下：

ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
backOffPolicy.setInitialInterval(200L);
backOffPolicy.setMultiplier(2);
backOffPolicy.setMaxInterval(5000L);

RetryTemplate retryTemplate = new RetryTemplate();
retryTemplate.setRetryPolicy((new SimpleRetryPolicy(MAX_VALUE)));
retryTemplate.setBackOffPolicy(backOffPolicy);
retryTemplate.registerListener(new RetryListenerSupport() {
    @Override
    public <T, E extends Throwable> void onError(RetryContext ctx, RetryCallback<T, E> callback, Throwable e) {
        log.error("Caught {} due to {} (count = {})", e.getClass().getSimpleName(), e.getMessage(), ctx.getRetryCount(), e);
    }
});
StatelessRetryOperationsInterceptorFactoryBean bean = new StatelessRetryOperationsInterceptorFactoryBean();
bean.setRetryOperations(retryTemplate);
bean.setMessageRecoverer(new ImmediateRequeueMessageRecoverer());
return bean.getObject();

[my-service-97c696799-6xs26] org.springframework.amqp.AmqpRejectAndDontRequeueException: Error Handler converted exception to fatal
[my-service-97c696799-6xs26]    at org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler.handleError(ConditionalRejectingErrorHandler.java:146)
[my-service-97c696799-6xs26]    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeErrorHandler(AbstractMessageListenerContainer.java:1436)
[my-service-97c696799-6xs26]    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.handleListenerException(AbstractMessageListenerContainer.java:1720)
[my-service-97c696799-6xs26]    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1495)
[my-service-97c696799-6xs26]    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:967)
[my-service-97c696799-6xs26]    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:913)
[my-service-97c696799-6xs26]    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83)
[my-service-97c696799-6xs26]    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1288)
[my-service-97c696799-6xs26]    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1194)
[my-service-97c696799-6xs26]    at java.base/java.lang.Thread.run(Thread.java:831)
[my-service-97c696799-6xs26] Caused by: org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener threw exception
[my-service-97c696799-6xs26]    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:1746)
[my-service-97c696799-6xs26]    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1636)
[my-service-97c696799-6xs26]    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1551)
[my-service-97c696799-6xs26]    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1539)
[my-service-97c696799-6xs26]    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1530)
[my-service-97c696799-6xs26]    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1474)
[my-service-97c696799-6xs26]    ... 6 common frames omitted
[my-service-97c696799-6xs26] Caused by: org.springframework.amqp.support.converter.MessageConversionException: Don't know how to convert (Body:'{ "yo" : "MTV Raps" }' MessageProperties [headers={content_type=application/json}, contentType=application/json, contentLength=0, receivedDeliveryMode=NON_PERSISTENT, redelivered=false, receivedExchange=, receivedRoutingKey=myservice.routingkey, deliveryTag=1, consumerTag=amq.ctag-9De2w0uuQxnve_9k6HZ7tw, consumerQueue=myservice.myqueue]) to an object because no event type was found
[my-service-97c696799-6xs26]    at com.mycompany.RabbitMQEventMessageConverter.fromMessage(RabbitMQEventMessageConverter.java:47)
[my-service-97c696799-6xs26]    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.convertPayload(AmqpInboundChannelAdapter.java:361)
[my-service-97c696799-6xs26]    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.createMessageFromAmqp(AmqpInboundChannelAdapter.java:342)
[my-service-97c696799-6xs26]    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.createAndSend(AmqpInboundChannelAdapter.java:334)
[my-service-97c696799-6xs26]    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:299)
[my-service-97c696799-6xs26]    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1632)
[my-service-97c696799-6xs26]    ... 10 common frames omitted

有人知道我如何在spring集成调用“服务方法”之前重新排队/重试失败的消息吗？

我们使用Spring Integration 5.4.6和Spring Boot 2.4.6。

Answer 1

在创建消息之前执行转换。

转换错误通常被认为是致命的--重试没有任何意义，因为它会再次失败。

向入站适配器添加.errorchannel；它的下游流将获得一个errormessage来表示转换错误。