SpringBoot（15） - 消息（3） - AMQP（1）RabbitMQ

悠悠 2022-04-18 00:44 77阅读 0赞

参考：[https://docs.spring.io/spring-boot/docs/1.5.17.RELEASE/reference/htmlsingle/\#boot-features-amqp][https_docs.spring.io_spring-boot_docs_1.5.17.RELEASE_reference_htmlsingle_boot-features-amqp]

高级消息队列协议（AMQP）是一个平台中立、线级协议的面向消息的中间件。 Spring AMQP项目将核心Spring概念应用于基于AMQP的消息传递解决方案的开发。Spring Boot提供了几种通过RabbitMQ使用AMQP的便利，包括spring-boot-starter-amqp的'Starter'。

RabbitMQ的支持

1. 简介

RabbitMQ是一个基于AMQP协议的轻量级、可靠、可扩展且可移植的消息代理。 Spring通过RabbitMQ使用AMQP协议进行通信。

RabbitMQ可以通过spring.rabbitmq.\*属性配置。例如，可以在application.properties中声明以下配置：

spring.rabbitmq.host=localhost
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=admin
    spring.rabbitmq.password=secret

查看org.springframework.boot.autoconfigure.amqp.RabbitProperties获取更多支持的属性。

2. 发送消息

Spring自动配置了AmqpTemplate和AmqpAdmin，可以直接通过@Autowired直接注入。

import org.springframework.amqp.core.AmqpAdmin;
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    @Component
    public class MyBean {
        private final AmqpAdmin amqpAdmin;
        private final AmqpTemplate amqpTemplate;
    
        @Autowired
        public MyBean(AmqpAdmin amqpAdmin, AmqpTemplate amqpTemplate) {
            this.amqpAdmin = amqpAdmin;
            this.amqpTemplate = amqpTemplate;
        }
        // ...
    }

注：可以以同样的方式注入RabbitMessagingTemplate。假如注册了MessageConverter的bean，将会自动关联到自动配置的AmqpTemplate上。

如果需要，任何定义org.springframework.amqp.core.Queue的bean将自动用于在RabbitMQ实例上声明相应的队列。

可以在AmqpTemplate上启用重试以重试操作，例如在broker连接丢失的情况下。 默认情况下禁用重试。

3. 接收消息

当RabbitMQ基础结构存在时，通过@RabbitListener注解的bean创建监听器端点。如果未定义RabbitListenerContainerFactory，将会自动配置默认值。假如定义了MessageConverter或MessageRecoverer的bean，它们将会自动关联到默认的factory上。

下面的示例中创建了一个监听someQueue队列的端点：

@Component
    public class MyBean {
        @RabbitListener(queues = "someQueue")
        public void processMessage(String content) {
            // ...
        }
    }

注：查看@EnableRabbit注解的Java文档查询更多信息。

假如想要创建更多的RabbitListenerContainerFactory实例或覆盖默认的工厂实例，SpringBoot提供了一个SimpleRabbitListenerContainerFactoryConfigurer用来初始化SimpleRabbitListenerContainerFactory，初始化后的实例和SpringBoot自动配置的实例具有相同的配置。

下面的例子中，新定义了一个使用特定MessageConverter的工厂：

@Configuration
    static class RabbitConfiguration {
        @Bean
        public SimpleRabbitListenerContainerFactory myFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer) {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            configurer.configure(factory, connectionFactory);
            factory.setMessageConverter(myMessageConverter());
            return factory;
        }
    }

然后通过@RabbitListener注解的containerFactory属性指定该工厂：

@Component
    public class MyBean {
        @RabbitListener(queues = "someQueue", containerFactory="myFactory")
        public void processMessage(String content) {
            // ...
        }
    }

可以启用重试来处理监听器抛出异常的情况。 默认使用RejectAndDontRequeueRecoverer，但可以自定义MessageRecoverer。 当超过重试最大次数时，如果配置了broker，则该消息将被拒绝并被丢弃或路由到死信exchange。 默认情况下禁用重试。

提示：如果未启用重试且监听器抛出异常，则默认情况下将无限期地重试传递。 可以通过两种方式修改此行为;：

*  将defaultRequeueRejected属性设置为false，并尝试零重新传递;
 *  抛出一个AmqpRejectAndDontRequeueException来表示应该拒绝该消息。 这是启用重试并达到最大传递尝试时使用的机制。

[https_docs.spring.io_spring-boot_docs_1.5.17.RELEASE_reference_htmlsingle_boot-features-amqp]: https://docs.spring.io/spring-boot/docs/1.5.17.RELEASE/reference/htmlsingle/#boot-features-amqp