处理RabbitMQ Spring Boot应用程序中的异常

Posted 2021-05-03

我使用的是Spring Boot 1.4.1-RELEASE和RabbitMQ 3.2.3。我的Application类看起来像这样 -

@SpringBootApplication
@EnableAutoConfiguration
public class EventStoreMessageDeliveryApplication {

    public final static String queueName = "customer.default.queue"; // spring-boot

    @Bean
    Queue queue() {
        return new Queue(queueName, true);
    }

    @Bean
    FanoutExchange exchange() {

        return new FanoutExchange("customer.events.fanout.exchange", true, false); // spring-boot-exchange
    }

    @Bean
    Binding binding() {

        return new Binding(queueName, Binding.DestinationType.QUEUE, "customer.events.fanout.exchange", "*.*", null);
    }

    @Bean
    public ConnectionFactory connectionFactory() {

        CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
        connectionFactory.setPublisherConfirms(true);

        return connectionFactory;
    }

    @Bean
    SimpleMessageListenerContainer container(MessageListenerAdapter listenerAdapter) {

        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory());
        container.setQueueNames(queueName);
        container.setMessageListener(listenerAdapter);
        container.setRecoveryBackOff(new ExponentialBackOff(3000, 2));
        return container;
    }

    @Bean
    MessageListenerAdapter listenerAdapter(Receiver receiver) {

        return new MessageListenerAdapter(receiver, "receiveMessage");
    }

    public static void main(String[] args) throws InterruptedException {

        SpringApplication.run(EventStoreMessageDeliveryApplication.class, args);
    }
}

我的听众课看起来像 -

@Component
public class Receiver {

    private CountDownLatch latch = new CountDownLatch(1);

    public void receiveMessage(String message) {

        System.out.println("Received <" + message + ">");

            // do something

        latch.countDown();
    }

    public CountDownLatch getLatch() {

        return latch;
    }

}

我想处理像经纪人关闭时可能出现的连接拒绝等异常。我该如何处理这些例外？我不知道在哪里可以获得异常的句柄。

答案

您可以创建SimpleRabbitListenerContainerFactory。这基本上是RabbitConnectionFactory事件的监听器。

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory());
    factory.setErrorHandler(rabbitErrorHandler());
    return factory;
}

rabbitErrorHandler()可以返回一个实施org.springframework.util.ErrorHandler的bean。

参考docs

另一答案

我有一个建议，它可以解决。由于您希望对RabbitMQ代理拒绝连接，因此客户端可以捕获异常。 在你的例子中，看起来像是来自SpringIO docs的那个，我假设你可以在main中进行异常处理（不推荐）：

@Component
    public class Runner implements CommandLineRunner {

        private final RabbitTemplate rabbitTemplate;
        private final Receiver receiver;

        public Runner(Receiver receiver, RabbitTemplate rabbitTemplate) {
            this.receiver = receiver;
            this.rabbitTemplate = rabbitTemplate;
        }

        @Override
        public void run(String... args) throws Exception {
            System.out.println("Sending message...");
            try {
                rabbitTemplate.convertAndSend(Application.topicExchangeName, "foo.bar.baz", "Hello from RabbitMQ!");
                receiver.getLatch().await(10000, TimeUnit.MILLISECONDS);
            }catch(AmqpException the_exception) {
                System.out.printl("Connection refused. Problem thrown when trying to connecto the RabbitMQ");
            }            
         }

    }

AmqpException来自convertAndSend()方法的文档，如果出现问题，它会被抛出。在这里，您可以捕获自己的自定义消息。 我希望这是你正在寻找的东西，或者至少引导你到达正确的目的地。

/一个