springboot与消息—使用、原理、整合RabbitMQ

怼烎@ 2022-05-10 02:44 226阅读 0赞

应用场景：异步处理、应用解耦、流量削峰

消息队列主要有两种形式的目的地：队列（queue）：点对点消息通信（point-to-point）；主题（topic）：发布（publish）/订阅（subscribe）消息通信。

JMS（Java Message Service）JAVA消息服务：基于JVM消息代理的规范。ActiveMQ、HornetMQ是JMS实现。

AMQP（Advanced Message Queuing Protocol）：高级消息队列协议，也是一个消息代理的规范，兼容JMS；RabbitMQ是AMQP的实现。

Spring Boot自动配置：JmsAutoConfiguration、RabbitAutoConfiguration。比较：

<table> 
 <tbody> 
 <tr> 
 <td style="width:142px;">&nbsp;</td> 
 <td style="width:372px;">JMS</td> 
 <td style="width:335px;">AMQP</td> 
 </tr> 
 <tr> 
 <td style="width:142px;">定义</td> 
 <td style="width:372px;">Java api</td> 
 <td style="width:335px;">网络线级协议</td> 
 </tr> 
 <tr> 
 <td style="width:142px;">跨语言</td> 
 <td style="width:372px;">否</td> 
 <td style="width:335px;">是</td> 
 </tr> 
 <tr> 
 <td style="width:142px;">跨平台</td> 
 <td style="width:372px;">否</td> 
 <td style="width:335px;">是</td> 
 </tr> 
 <tr> 
 <td style="width:142px;">Model</td> 
 <td style="width:372px;"> 提供两种消息模型： 1、Peer-2-Peer（点对点） 2、Pub/sub（发布订阅模型） </td> 
 <td style="width:335px;"> 提供了五种消息模型： 1、direct exchange（点对点） 2、fanout exchange 3、topic change 4、headers exchange 5、system exchange 本质来讲，后四种和JMS的pub/sub模型没有太大差别，仅是在路由机制上做了更详细的划分； </td> 
 </tr> 
 <tr> 
 <td style="width:142px;">支持消息类型</td> 
 <td style="width:372px;">多种消息类型：TextMessage、MapMassage、BytesMessage、StreamMessage、ObjectMassage、Message</td> 
 <td style="width:335px;"> byte[] 当实际应用时，有复杂的消息，可以将消息序列化后发送。 </td> 
 </tr> 
 <tr> 
 <td style="width:142px;">综发评价</td> 
 <td style="width:372px;">JMS定义了JavaAPI层面的标准；在java体系中，多个client均可通过JMS进行交互，不需要应用修改代码，但是其对跨平台的支持较差。</td> 
 <td style="width:335px;">AMQP定义了wire level层的协议标准；天然具有跨平台、跨语言特性。</td> 
 </tr> 
 </tbody> 
</table>

1、配置整合RabbitMQ

RabbitMQ是一个由erlang开发的AMQP(Advanved Message Queue Protocol)的开源实现。

AMQP 中消息的路由过程：AMQP 中增加了 Exchange （交换器）和 Binding（绑定的路由规则） 的角色。生产者把消息发布到 Exchange 上，消息最终到达队列并被消费者接收，而 Binding 决定交换器的消息应该发送到那个队列。

a、安装Erlang：Rabbit MQ 是建立在强大的Erlang OTP平台上，因此安装Rabbit MQ的前提是安装Erlang。

地址：http://www.erlang.org/downloads

b、安装Rabbit MQ

地址：http://www.rabbitmq.com/download.html

c、进入到sbin目录开启rabbitmq\_management插件：rabbitmq\_management是管理后台的插件、开启这个插件才能通过浏览器访问登录页面。

rabbitmq-plugins enable rabbitmq\_management

![70][]

d、启动Rabbit MQ，在浏览器中打开，登录管理界面：rabbitmq-server start

![70 1][]

e、测试交换机制

![70 2][]

创建交换器exchange.direct、exchange.fanout、exchange.topic

![70 3][]

添加消息队列：msgtest、msgtest.news、msgtest.emps、msgtest1.news；例：

![70 4][]

关联绑定交换器，例:

![70 5][]

![70 6][]

给交换器发送消息，来测试哪个队列接收到消息

![70 7][]

对于direct交换器是完全匹配路由键的，所以只有相匹配的msgtest队列才能收到消息。

![70 8][]

2、springboot整合测试RabbitMQ

a、引入 spring-boot-starter-amqp

<dependency>
 	<groupId>org.springframework.boot</groupId>
 	 <artifactId>spring-boot-starter-amqp</artifactId>
 </dependency>

b、配置RabbitMQ基本信息：

#RabbitMQ主机地址、用户名、密码、端口，默认localhost、guest、guest、5672
    spring.rabbitmq.host=127.0.0.1
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest
    spring.rabbitmq.port=5672

c、测试：两种发送方式：rabbitTemplate.send(交换器,路由键,消息);rabbitTemplate.convertAndSend(交换器,路由键,数据对象);

@RunWith(SpringRunner.class)
 @SpringBootTest
 public class SpringbootApplicationTests implements Serializable {
 @Autowired
 	RabbitTemplate rabbitTemplate;
 
 /**
 	 *单播（点对点）
 	 */
 	@Test
 	public void testRabbit() {
 		//rabbitTemplate.convertAndSend("exchange.direct","msgtest.news",new Book("西游记","吴承恩"));
 		//rabbitTemplate.send(交换器,路由键,消息);
 		//message需要自己构造一个(package org.springframework.amqp.core;使用核心包中的Message)；定义消息体内容和消息头
 		//rabbitTemplate.send(exchange,routekey,message);
 
 		//转化并发送，只需要传入要发送的对象，自动序列化发送给rabbitmq；
 		//rabbitTemplate.convertAndSend(交换器,路由键,数据对象);
 		//object：数据对象将默认当成消息体，并自动转化
 		Map<String,Object> map = new HashMap<>();
 		map.put("msg","这是第一个消息！");
 		map.put("data", Arrays.asList("helloworld",123,true));
 		rabbitTemplate.convertAndSend("exchange.direct","msgtest.news",map);
 	}
 }

查看结果，默认使用了x-java-serialized-object方式序列化对象（已发送成功）

![70 9][]

d、接收消息：rabbitTemplate.receive("消息队列");rabbitTemplate.receiveAndConvert("消息队列")

@Test
    public void receive(){
    	//接收消息 返回message，有消息头及消息体
    	//rabbitTemplate.receive("消息队列");
    	//接收并转化成对象返回
    	Object o = rabbitTemplate.receiveAndConvert("msgtest.news");
    	System.out.println(o.getClass());//类型是HashMap
    	System.out.println(o);
    }

接收到消息之后，刷新消息队列，该消息将不存在了

![70 10][]

e、如何将数据自动转为json发送出去

![70 11][]

使用Jackson2JsonMessageConverter

![70 12][]

自定义messageConverter后，根据自动配置将使用自定义的messageConverter

![70 13][]

自定义messageConverter

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.amqp.support.converter.MessageConverter;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class MyRMQConfig {
        @Bean
        public MessageConverter messageConverter(){
            return  new Jackson2JsonMessageConverter();
        }
    }

重新发送消息，此时获取的是json格式数据

![70 14][]

3、RabbitMQ自动配置原理

自动配置类：RabbitAutoConfiguration：

a、自动配置了连接工厂ConnectionFactory

b、RabbitProperties 封装了 RabbitMQ的配置

c、RabbitTemplate ：给RabbitMQ发送和接收消息

d、AmqpAdmin ：RabbitMQ的系统管理功能组件（申明一个队列，创建一个交换器等）

[70]: /images/20220510/9c097bc28b774cb1825cb599b64079ac.png
[70 1]: /images/20220510/f90770ab72e941d4afa39c64c86c5b9f.png
[70 2]: /images/20220510/7b2a120db7e44e958a3727d7f977e03e.png
[70 3]: /images/20220510/ca3cea2125dd4ff897650a8d50d4c119.png
[70 4]: /images/20220510/0ce76f3769214225b61b4818e67c1185.png
[70 5]: /images/20220510/9d3f699eb7324609a66b81f83e5f515f.png
[70 6]: /images/20220510/f80fc65604b34eb2a340108d57923be8.png
[70 7]: /images/20220510/2ddaffe7dfab41659ca68d7763a1557e.png
[70 8]: /images/20220510/9ba3cc681f794166b8ddedb661cf5c2f.png
[70 9]: /images/20220510/54c3863282734785a8df6d6cd516353c.png
[70 10]: /images/20220510/4621356464d843dfaf632d3861a296bb.png
[70 11]: /images/20220510/1e45cdff5b7f46d89cbd046b7558c292.png
[70 12]: /images/20220510/073b809d454e44ed97da45a76cf2e861.png
[70 13]: /images/20220510/da0b00f133324f8ea6b55c2abda00c5c.png
[70 14]: /images/20220510/c9afa387929440038524f96a2f918ca8.png