4

我正在使用 Spring AMQP 发送消息并能够对“自定义”异常执行重试。假设我有一个抛出自定义异常“EventException”的接收器，为此，我希望有n次重试（在我们的示例 5 中）。在重试之间，我也希望有 5 秒的延迟。这是我的源代码：

@SpringBootApplication
public class DemoApplication implements CommandLineRunner {

    final static String queueName = "testing-queue";

    @Autowired
    AnnotationConfigApplicationContext context;

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Bean
    Queue queue() {
        Map<String, Object> arguments = new HashMap<String, Object>();
        arguments.put("x-dead-letter-exchange", "dead-letter-exchange");
        Queue queue = new Queue(queueName, true, false, false, arguments);
        return queue;
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange("testing-exchange");
    }

    @Bean
    Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(queueName);
    }

    @Bean
    Queue deadLetterQueue() {
        return new Queue("dead-letter-queue", true);
    }

    @Bean
    FanoutExchange deadLetterExchange() {
        return new FanoutExchange("dead-letter-exchange");
    }

    @Bean
    Binding deadLetterBinding(Queue deadLetterQueue, FanoutExchange deadLetterExchange) {
        return BindingBuilder.bind(deadLetterQueue).to(deadLetterExchange);
    }

    @Bean
    ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = 
                new CachingConnectionFactory("localhost");

        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");

        return connectionFactory;
    }

    @Bean
    SimpleMessageListenerContainer container(
            ConnectionFactory connectionFactory,
            MessageListenerAdapter listenerAdapter,
            RetryOperationsInterceptor interceptor) {

        Advice[] adviceChain = { interceptor };

        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(queueName);
        container.setAdviceChain(adviceChain);
        container.setMessageListener(listenerAdapter);

        return container;
    }

    @Bean
    Receiver receiver() {
        return new Receiver();
    }

    @Bean
    MessageListenerAdapter listenerAdapter(Receiver receiver) {
        MessageListenerAdapter adapter = 
                new MessageListenerAdapter(receiver, "receiveMessage");

        return adapter;
    }

    @Bean
    RetryOperations retryTemplate() {
         Map<Class<? extends Throwable>, Boolean> retryableExceptions = 
                 new HashMap<Class<? extends Throwable>, Boolean>();
        retryableExceptions.put(EventException.class, false);

        FixedBackOffPolicy backoffPolicy = new FixedBackOffPolicy();
        backoffPolicy.setBackOffPeriod(5000);

        RetryTemplate retryTemplate = new RetryTemplate();
        retryTemplate.setBackOffPolicy(backoffPolicy);
        retryTemplate.setRetryPolicy(new SimpleRetryPolicy(5, retryableExceptions));

        return retryTemplate;
    }

    @Bean
    RetryOperationsInterceptor interceptor(RetryOperations retryTemplate) {
        RetryOperationsInterceptor interceptor = new RetryOperationsInterceptor();
        interceptor.setRecoverer(new CustomMessageRecover());
        interceptor.setRetryOperations(retryTemplate);

        return interceptor;
//      return RetryInterceptorBuilder
//              .stateless()
//              //.retryOperations(retryTemplate)
//              .maxAttempts(5)
//              .recoverer(new CustomMessageRecover()).build();
    }

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

    @Override
    public void run(String... args) throws Exception {
        System.out.println("Sending message...");
        rabbitTemplate.convertAndSend(queueName, "Hello from RabbitMQ!");
        context.close();
    }

    public class Receiver {

        public void receiveMessage(String message) throws Exception {
            System.out.println("!!!!!!!!Message has been recieved!!!!!!");
            throw new EventException("TESTING");
        }
    }

    public class CustomMessageRecover implements MethodInvocationRecoverer<Void> {

        @Override
        public Void recover(Object[] args, Throwable cause) {
            System.out.println("IN THE RECOVER ZONE!!!");
            throw new AmqpRejectAndDontRequeueException(cause);
        }
    }

    class EventException extends Exception {
        private static final long serialVersionUID = 1L;

        public EventException() {}

        public EventException(String message) {
            super(message);
        }
    }
}

现在在代码中，如您所见，我正在使用RetryOperationsInterceptor来拦截并检查它被抛出的异常类型，并基于此，决定是否重试，以及之间的延迟重试。

为此，我设置了RetryTemplate Bean 的backoffPolicy和retryPolicy并将其注入到 RetryOperationsInterceptor 中。

如果有人可以帮助我并告诉我为什么重试和重试之间的延迟不起作用，我将不胜感激。我的消息将直接发送到死信交换，而不会发生重试和延迟。

谢谢你！

4

1 回答 1

1

你的问题在这里：

retryableExceptions.put(EventException.class, false);

请找到SimpleRetryPolicy代码：

public boolean canRetry(RetryContext context) {
    Throwable t = context.getLastThrowable();
    return (t == null || retryForException(t)) && context.getRetryCount() < maxAttempts;
}

并进一步：

private boolean retryForException(Throwable ex) {
    return retryableClassifier.classify(ex);
}

false由于您为您的指定 a EventException，因此不会是retryable。因此，任何重试和退避。

于 2015-04-11T11:52:02.770 回答