Spring rabbit 重试传递被拒绝的消息..可以吗? [英] Spring rabbit retries to deliver rejected message..is it OK?

问题描述

我有以下配置

spring.rabbitmq.listener.prefetch=1
spring.rabbitmq.listener.concurrency=1
spring.rabbitmq.listener.retry.enabled=true
spring.rabbitmq.listener.retry.max-attempts=3
spring.rabbitmq.listener.retry.max-interval=1000
spring.rabbitmq.listener.default-requeue-rejected=false //I have also changed it to true but the same behavior still happens

在我的监听器中，我抛出异常 AmqpRejectAndDontRequeueException 拒绝消息并强制兔子不要尝试重新传递它...但是兔子将它重新传递了 3 次最后将其路由到死信队列.

and in my listener I throw the exception AmqpRejectAndDontRequeueException to reject the message and enforce rabbit not to try to redeliver it...But rabbit redilvers it for 3 times then finally route it to dead letter queue.

这是根据我提供的配置的标准行为还是我错过了什么?

Is that the standard behavior according to my provided configuration or do I miss something?

推荐答案

您必须将重试策略配置为不为该异常重试.

You have to configure the retry policy to not retry for that exception.

你不能用属性来做到这一点，你必须自己配置重试建议.

You can't do that with properties, you have to configure the retry advice yourself.

如果您需要帮助，我稍后会发布一个示例.

I'll post an example later if you need help with that.

requeue-rejected 处于容器级别(在堆栈上重试下方).

requeue-rejected is at the container level (below retry on the stack).

编辑

@SpringBootApplication
public class So39853762Application {

    public static void main(String[] args) throws Exception {
        ConfigurableApplicationContext context = SpringApplication.run(So39853762Application.class, args);
        Thread.sleep(60000);
        context.close();
    }

    @RabbitListener(queues = "foo")
    public void foo(String foo) {
        System.out.println(foo);
        if ("foo".equals(foo)) {
            throw new AmqpRejectAndDontRequeueException("foo"); // won't be retried.
        }
        else {
            throw new IllegalStateException("bar"); // will be retried
        }
    }

    @Bean
    public ListenerRetryAdviceCustomizer retryCustomizer(SimpleRabbitListenerContainerFactory containerFactory,
            RabbitProperties rabbitPropeties) {
        return new ListenerRetryAdviceCustomizer(containerFactory, rabbitPropeties);
    }

    public static class ListenerRetryAdviceCustomizer implements InitializingBean {

        private final SimpleRabbitListenerContainerFactory containerFactory;

        private final RabbitProperties rabbitPropeties;

        public ListenerRetryAdviceCustomizer(SimpleRabbitListenerContainerFactory containerFactory,
                RabbitProperties rabbitPropeties) {
            this.containerFactory = containerFactory;
            this.rabbitPropeties = rabbitPropeties;
        }

        @Override
        public void afterPropertiesSet() throws Exception {
            ListenerRetry retryConfig = this.rabbitPropeties.getListener().getRetry();
            if (retryConfig.isEnabled()) {
                RetryInterceptorBuilder<?> builder = (retryConfig.isStateless()
                        ? RetryInterceptorBuilder.stateless()
                        : RetryInterceptorBuilder.stateful());
                Map<Class<? extends Throwable>, Boolean> retryableExceptions = new HashMap<>();
                retryableExceptions.put(AmqpRejectAndDontRequeueException.class, false);
                retryableExceptions.put(IllegalStateException.class, true);
                SimpleRetryPolicy policy =
                        new SimpleRetryPolicy(retryConfig.getMaxAttempts(), retryableExceptions, true);
                ExponentialBackOffPolicy backOff = new ExponentialBackOffPolicy();
                backOff.setInitialInterval(retryConfig.getInitialInterval());
                backOff.setMultiplier(retryConfig.getMultiplier());
                backOff.setMaxInterval(retryConfig.getMaxInterval());
                builder.retryPolicy(policy)
                    .backOffPolicy(backOff)
                    .recoverer(new RejectAndDontRequeueRecoverer());
                this.containerFactory.setAdviceChain(builder.build());
            }
        }

    }

}

注意:您当前无法配置策略以重试所有异常，除了"这个异常 - 您必须对要重试的所有异常进行分类(并且它们不能是 AmqpRejectAndDontRequeueException).我已经打开了一个问题来支持这个.

NOTE: You cannot currently configure the policy to retry all exceptions, "except" this one - you have to classify all exceptions you want retried (and they can't be a superclass of AmqpRejectAndDontRequeueException). I have opened an issue to support this.

这篇关于Spring rabbit 重试传递被拒绝的消息..可以吗?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！