SpringBoot2.x整合RabbitMQ

偏执的太偏执、 2021-12-11 21:21 388阅读 0赞

在之前的博客中[https://blog.csdn.net/zhoujian\_Liu/article/details/95197408][https_blog.csdn.net_zhoujian_Liu_article_details_95197408]，讲解了一下RabbitMQ的核心概念及如何使用web页面手动创建队列、交换器及绑定规则，在本文中将介绍SpringBoot与RabbitMQ的整合方法，使用Java代码的方式创建交换器、队列及绑定关系，并使用自定义配置替代RabbitMQ中默认JDK序列化方式，将结果转为JSON 保存在队列中。

### **1、环境搭建：创建一个SpringBoot项目** ###

**1.1、pom.xml**

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.1.2.RELEASE</version>
            <relativePath/> 
        </parent>
        <groupId>com.lzj</groupId>
        <artifactId>springboot-rabbitmq</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>springboot-rabbitmq</name>
        <description>Demo project for Spring Boot</description>
    
        <properties>
            <java.version>1.8</java.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <optional>true</optional>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    
    </project>

**1.2、application.properties**

spring.rabbitmq.host=127.0.0.1
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest
    spring.rabbitmq.port=5672

**1.3 启动类**

@EnableRabbit //开启rabbitmq注解
    @SpringBootApplication
    public class SpringbootRabbitmqApplication {
        public static void main(String[] args) {
            SpringApplication.run(SpringbootRabbitmqApplication.class, args);
        }
    }

使用@EnableRabbit注解开启SpringBoot对RabblitMQ的注解支持。

至此基本环境搭建完毕，启动程序，可以看到已连接到RabbitMQ.

![20190709175721196.png][]

### 2、编写单元测试 ###

由于RabbitMQ的自动配置RabbitAutoConfiguration给我们创建了**配置连接工厂、RabbitTemplate、AmqpAdmin**等，我们可以直接在代码中**注入RabbitTemplate**向RabbitMQ中发送或接受消息，也可以**注入AmqpAdmin**用以创建交换器、队列及绑定规则。

**2.1 使用AmqpAdmin 创建队列、交换器、绑定规则**

代码如下：

@Test
        public void createExchangeAndQueue() {
            //构建exchange，这里简单的使用一个参数的构造器：参数为：交换器名称；还有其他的构造器可以设置持久化及自动删除等。
            amqpAdmin.declareExchange(new DirectExchange("test.direct"));
    
            //构建queue，第一个参数为队列名，第二个参数为是否持久化
            amqpAdmin.declareQueue(new Queue("test.queue", true));
    
            //构建绑定规则：第一个参数为绑定队列名、第二个参数为绑定类型、第三个参数为绑定交换器名、第四个参数为路由键、第五个参数：参数列表
            amqpAdmin.declareBinding(new Binding("test.queue", 
            Binding.DestinationType.QUEUE,"test.direct" , "test.lzj", null));
        }

此外，共有如下五种类型的交换器，可以根据自己的需求选择：

![20190709180542472.png][]

运行测试代码：可以看到已经创建了一个名为test.queue的队列和test.direct的交换器，并且绑定了test.lzj为路由键。

![20190709182411552.png][]

![20190709182318275.png][]

**2.2 单播，点对点模式**

/**
         * 单播-点对点
         */
        @Test
        public void sendMsgToOne() {
            //参数依次为：exchange名称、routingKey、消息
            Map<String, Object> maps = new HashMap<>();
            maps.put("message", "正常");
            maps.put("status", 200);
            maps.put("data", Arrays.asList("张三", "李四"));
            rabbitTemplate.convertAndSend("test.direct", "test.lzj",  maps);
        }
    
    
        //从queue中取出数据
        @Test
        public void receive() {
            Object o = rabbitTemplate.receiveAndConvert("test.queue");
            System.out.println(o.getClass());
            System.out.println(o);
        }

在代码中把消息交给了test.direct交换器，并设置的路由键是test-lzj，于是交换器会拿着test-lzj路由键进行全局匹配，会将这个消息转发到与之绑定的test-queue中。

运行代码：可以看到test.queue已经有了一条消息。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3VqaWFuX0xpdQ_size_16_color_FFFFFF_t_70][]

运行从queue中取数据的代码：可以看到从队列中成功获取了数据。

![20190709185458294.png][]

但是通过web界面查看里面的消息，发现都读不懂，这是因为默认序列化机制采用了SimpleMessageConverter,而SimpleMessageConverter底层还是用了JDK的序列化机制。

源码：

content = SerializationUtils.deserialize(this.createObjectInputStream(new ByteArrayInputStream(message.getBody()), this.codebaseUrl));

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3VqaWFuX0xpdQ_size_16_color_FFFFFF_t_70 1][]

为了是开发者方便阅读消息内容，可以自己配置MessageConverter。（***注意若要发送自定义对象必须配置MessageConverter，因为默认的SimpleMessageConverter只支持string、byte\[\]和序列化后的数据，若没有配置报错：java.lang.IllegalArgumentException: SimpleMessageConverter only supports String, byte\[\] and Serializable payloads***）

@Configuration
    public class AMQPConfig {
    
        //配置messageConverter，在RabbitMQ中以JSON格式存储数据,不配置默认使用Java的序列化机制
        @Bean
        public MessageConverter messageConverter() {
            return new Jackson2JsonMessageConverter();
        }
    }

此外为了测试，编写有个user对象，代码如下：

@Data
    @NoArgsConstructor
    @AllArgsConstructor
    public class User {
    
        private String name;
        private Integer age;
    }

测试代码如下：

@Test
        public void sendMsgToOne() {
           rabbitTemplate.convertAndSend("test.direct", "test.lzj", new User("张三", 18));
        }
    
        @Test
        public void receive() {
            User user = (User) rabbitTemplate.receiveAndConvert("test-queue");
            System.out.println(user);
        }

通过web查看结果如下：数据以JSON串进行存储。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3VqaWFuX0xpdQ_size_16_color_FFFFFF_t_70 2][]

接收数据测试代码运行结果：

![2019070919352450.png][]

**2.3 广播**

**创建fanout交换器、两个队列及绑定规则**

@Test
        public void createExchangeAndQueue() {
            //2、构建广播类型 exchange
            amqpAdmin.declareExchange(new FanoutExchange("test.fanout"));
            amqpAdmin.declareQueue(new Queue("test.queue1"));
            amqpAdmin.declareQueue(new Queue("test.queue2"));
            amqpAdmin.declareBinding(new Binding("test.queue1", Binding.DestinationType.QUEUE, "test.fanout", "", null));
            amqpAdmin.declareBinding(new Binding("test.queue2", Binding.DestinationType.QUEUE, "test.fanout", "", null));
        }

结果如下：

![20190709194322696.png][]

测试代码：

@Test
        public void sendMsgToAll() {
            rabbitTemplate.convertAndSend("test.fanout", "",  new User("张三", 18));
        }

运行后的结果：

![20190709194456224.png][]

### 3、监听方式接收队列中的数据 ###

@Service
    public class UserService {
    
        @RabbitListener(queues = "test.queue")
        public void receive(User user) {
            System.out.println(user);
        }
    
        @RabbitListener(queues = "test.queue")
        public void receiveMessage(Message message) {
            System.out.println("receiveMessage:" + message.getBody());
            System.out.println("receiveMessage:" + message.getMessageProperties());
        }
    }

使用@RabblitListener（监听队列名称）。方法的参数可以为User类型，也可以为Message类型，Message类型可以获取到数据头部信息。***注意要在启动类型添加@EnableRabbit，@RabbitListener才能生效！***

[Github代码][Github]

[https_blog.csdn.net_zhoujian_Liu_article_details_95197408]: https://blog.csdn.net/zhoujian_Liu/article/details/95197408
[20190709175721196.png]: /images/20211211/83159d0f4675411ba1f8ef9cb238b5bf.png
[20190709180542472.png]: /images/20211211/f61b9e9e28d54a9aa3a2b6cd1ba82d12.png
[20190709182411552.png]: /images/20211211/49cedf12ef8c46888470e28a0376b491.png
[20190709182318275.png]: /images/20211211/d827da8795e14aa7bdcbbd69bc1e097e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3VqaWFuX0xpdQ_size_16_color_FFFFFF_t_70]: /images/20211211/1d3dc49fd6c64efe91d740f3b3b50090.png
[20190709185458294.png]: /images/20211211/7d8c9b35607e4682a2182bf5c75015c3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3VqaWFuX0xpdQ_size_16_color_FFFFFF_t_70 1]: /images/20211211/1c6ab069391e46dba50794e487c745d4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3VqaWFuX0xpdQ_size_16_color_FFFFFF_t_70 2]: /images/20211211/03cbc0049c6e405984347716ef5e0031.png
[2019070919352450.png]: /images/20211211/2389d79fcf274f74b516a5ec20f2b899.png
[20190709194322696.png]: /images/20211211/5259e159b17c4cbfb0977097170eb3ec.png
[20190709194456224.png]: /images/20211211/9694fc493df84df98a599540ac4abd1a.png
[Github]: https://github.com/liuzhoujian/springboot-rabbitmq