RabbitMq 底子

文章目录

• 一、初识 MQ
  
• 
  
  
  
  • 1.1 同步调用：
    
  • 1.2 异步调用：
    
  
  
• 二、RabbitMQ
  
• 三、SpringAMQP
  
• 
  
  
  
  • 3.1 依赖和设置文件
    
  • 3.2 消息发送和汲取：
    
  • 
    
    
    
    • 3.2.1 消息发送：
      
    • 3.2.2 消息汲取：
      
    
    
  • 3.3 WorkQueues 模型：
    
  • 3.4 互换机范例：
    
  • 
    
    
    
    • 3.4.1 Fanout 互换机：
      
    • 3.4.2 Direct 互换机：
      
    • 3.4.3 Topic 互换机：
      
    
    
  • 3.5 声明互换机和队列：
    
  • 
    
    
    
    • 3.5.1 使用 API：
      
    • 
      
      
      
      • 3.5.1.1 fanout 示例：
        
      • 3.5.1.2 direct 示例：
        
      
      
    • 3.5.2 使用注解：
      
    • 
      
      
      
      • 3.5.2.1 fanout 示例:
        
      • 3.5.2.2 direct 示例：
        
      
      
    
    
  • 3.6 消息转化器：
    
  
  

一、初识 MQ

1.1 同步调用：

我们观察下，下面这个余额付出功能的流程图：

如果我们采用的是基于 OpenFeign 的同步调用，也就是说业务执行流程是这样的：

• 付出服务需要先调用用户服务完成余额扣减。
  
• 然后付出服务自己要更新付出流水单的状态。
  
• 然后付出服务调用生意业务服务，更新业务订单状态为已付出。
  

三个步调依次执行。
这其中就存在3个标题：

• 拓展性差：
  

我们目前的业务相对简朴，但是随着业务规模扩大，产物的功能也在不断完善。
在大多数电贸易务中，用户付出成功后都会以短信或者别的方式关照用户，告知付出成功。假如后期产物经理提出这样新的需求，你怎么办？是不是要在上述业务中再参加关照用户的业务？
某些电商项目中，还会有积分或金币的概念。假如产物经理提出需求，用户付出成功后，给用户以积分嘉奖或者返还金币，你怎么办？是不是要在上述业务中再参加积分业务、返还金币业务？
终极你的付出业务会越来越臃肿：

也就是说每次有新的需求，现有付出逻辑都要跟着变化，代码经常变动，不符合开闭原则，拓展性不好。

• 性能降落：
  

由于我们采用了同步调用，调用者需要等候服务提供者执行完返回效果后，才能继承向下执行，也就是说每次远程调用，调用者都是壅闭等候状态。终极整个业务的响应时长就是每次远程调用的执行时长之和：

假如每个微服务的执行时长都是 50ms，则终极整个业务的耗时可能高达 300ms，性能太差了。

• 级联失败：
  

由于我们是基于 OpenFeign 调用生意业务服务、关照服务。当生意业务服务、关照服务出现故障时，整个事件都会回滚，生意业务失败。
这实在就是同步调用的级联失败标题。
不能因为短信关照、更新订单状态失败而回滚整个事件（这些都不是付出服务的主线任务）。
而要解决这些标题，我们就必须用异步调用的方式来代替同步调用。
1.2 异步调用：

异步调用方式实在就是基于消息关照的方式，一样寻常包罗三个角色：

• 消息发送者：投递消息的人，就是原来的调用方。
  
• 消息Broker：管理、暂存、转发消息。
  
• 消息汲取者：汲取和处理惩罚消息的人，就是原来的服务提供方。
  

还是以余额付出业务为例：

除了扣减余额、更新付出流水单状态以外，别的调用逻辑全部取消。而是改为发送一条消息到 Broker。而相关的微服务都可以订阅消息关照，一旦消息到达 Broker，则会分发给每一个订阅了的微服务，处理惩罚各自的业务。
假如产物经理提出了新的需求，比如要在付出成功后更新用户积分。付出代码完全不用变更，而仅仅是让积分服务也订阅消息即可：

不管后期增长了多少消息订阅者，作为付出服务来讲，执行问扣减余额、更新付出流水状态后，发送消息即可。业务耗时仅仅是这三部分业务耗时，仅仅 100ms，大大提高了业务性能。
另外，不管是生意业务服务、关照服务，还是积分服务，他们的业务与付出关联度低。如今采用了异步调用，解除了耦合，他们即便执行过程中出现了故障，也不会影响到付出服务。
综上，异步调用的优势包括：

• 耦合度更低
  
• 性能更好
  
• 业务拓展性强
  
• 故障隔离，避免级联失败
  

当然，异步通讯也并非完美无缺，它存在下列缺点：

• 完全依赖于 Broker 的可靠性、安全性和性能
  
• 架构复杂，后期维护和调试贫苦
  

消息 Broker，目前常见的实现方案就是消息队列（MessageQueue），简称为 MQ。
下面我们就来进行 RabbitMQ 的学习。
二、RabbitMQ

RabbitMQ 是基于 Erlang 语言开发的开源消息通讯中间件，官网地址：
https://www.rabbitmq.com/
接下来，我们就学习它的根本概念和底子用法。
安装：
基于 Docker 来安装 RabbitMQ，使用下面的下令即可：

1. docker run

复制代码

-e RABBITMQ_DEFAULT_USER=itheima
-e RABBITMQ_DEFAULT_PASS=123321
-v mq-plugins:/plugins
–name mq
–hostname mq
-p 15672:15672
-p 5672:5672
–network hm-net
-d
rabbitmq:3.8-management
可以看到在安装下令中有两个映射的端口：

• 15672：RabbitMQ提供的管理控制台的端口
  
• 5672：RabbitMQ的消息发送处理惩罚接口
  

安装完成后，我们访问 http://ip地址:15672即可看到管理控制台。初次访问需要登录，默认的用户名和密码在设置文件中已经指定了。
登录后即可看到管理控制台总览页面：

RabbitMQ 对应的架构如图：

其中包罗几个概念：

• publisher：生产者。
  
• consumer：消费者。
  
• queue：队列，存储消息。生产者投递的消息会暂存在消息队列中，等候消费者处理惩罚。
  
• exchange：互换机，负责消息路由。生产者发送的消息由互换机决定投递到哪个队列。
  
• virtual host：假造主机，起到数据隔离的作用。每个假造主机相互独立，有各自的 exchange、queue。这个就像一个 mysql 有多个数据库每个数据库有自己的表。
  

在 MQ 的客户端也可以实现收发消息，比较简朴，所以这里跳过。
三、SpringAMQP

未来我们开发业务功能的时候，肯定不会在控制台收发消息，而是应该基于编程的方式。由于RabbitMQ采用了 AMQP 协议，因此它具备跨语言的特性。任何语言只要遵循 AMQP 协议收发消息，都可以与RabbitMQ交互。并且RabbitMQ官方也提供了各种不同语言的客户端。
但是，RabbitMQ 官方提供的 Java 客户端编码相对复杂，一样寻常生产环境下我们更多会结合 Spring 来使用。而 Spring 的官方刚好基于RabbitMQ 提供了这样一套消息收发的模板工具：SpringAMQP。并且还基于 SpringBoot 对实在现了主动装配，使用起来非常方便。
SpringAmqp 的官方地址：
https://spring.io/projects/spring-amqp
SpringAMQP 提供了三个功能：

• 主动声明队列、互换机及其绑定关系。
  
• 基于注解的监听器模式，异步汲取消息。
  
• 封装了 RabbitTemplate 工具，用于发送消息。
  

3.1 依赖和设置文件

要进行下面的操作，首先我们需要引入相关依赖：

1. <!--AMQP依赖，包含RabbitMQ-->
2. <dependency>
3.     <groupId>org.springframework.boot</groupId>
4.     <artifactId>spring-boot-starter-amqp</artifactId>
5. </dependency>

复制代码

其次进行设置文件：
在application.yml中添加设置：

1. spring:
2.   rabbitmq:
3.     host: xxxxx # 你的虚拟机IP
4.     port: 5672 # 端口
5.     virtual-host: /hmall # 虚拟主机
6.     username: hmall # 用户名
7.     password: 123 # 密码

复制代码

注意：岂论是消息的发送者还是消息的汲取者都需要进行上面的设置。
3.2 消息发送和汲取：

3.2.1 消息发送：

使用RabbitTemplate实现消息发送，RabbitTemplate 在我们引入依赖和设置完文件启动项目就主动创建了，我们直接进行依赖注入即可。
下面的代码中，没有实现互换机，因为比较复杂，我们下面讲，这里实在就是感受 RabbitTemplate 的使用。队列需要我们在 MQ 控制台手动创建。

1. public class SpringAmqpTest {
3.     @Autowired
4.     private RabbitTemplate rabbitTemplate;
6.     public void testSimpleQueue() {
7.         // 队列名称
8.         String queueName = "simple.queue";
9.         // 消息
10.         String message = "hello, spring amqp!";
11.         // 发送消息
12.         rabbitTemplate.convertAndSend(queueName, message);
13.     }
14. }

复制代码

3.2.2 消息汲取：

1. @Component
2. public class SpringRabbitListener {
3.     // 利用RabbitListener来声明要监听的队列信息
4.     // 将来一旦监听的队列中有了消息，就会推送给当前服务，调用当前方法，处理消息。
5.     @RabbitListener(queues = "simple.queue")
6.     public void listenSimpleQueueMessage(String msg) throws InterruptedException {
7.         System.out.println("spring 消费者接收到消息：【" + msg + "】");
8.     }
9. }

复制代码

注意：消息汲取的方法参数范例要和消息的发送的参数范例一致。例如上面都是 String 范例。
3.3 WorkQueues 模型：

Work queues，任务模型。简朴来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。

当消息处理惩罚比较耗时的时候，可能生产消息的速率会远远大于消息的消费速率。长此以往，消息就会堆积越来越多，无法实时处理惩罚。
此时就可以使用work 模型，多个消费者共同处理惩罚消息处理惩罚，同一条消息只会被一个消费者处理惩罚，这样，消息处理惩罚的速率就能大大提高了。
不过在 RabbitMQ 中该模型还存在一个标题：消息是均匀分配给每个消费者，并没有思量到消费者的处理惩罚本领。
这显然是不公道的。
改进：
在spring中有一个简朴的设置，可以解决这个标题，在 application.yml 文件中添加设置：

1. spring:
2.   rabbitmq:
3.     listener:
4.       simple:
5.         prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

复制代码

这样就能实现能者多劳的效果，充分使用了每一个消费者的处理惩罚本领，可以有效避免消息积蓄标题。
总结：
Work模型的使用：

• 多个消费者绑定到一个队列，同一条消息只会被一个消费者处理惩罚。
  
• 通过设置prefetch来控制消费者预取的消息数量。
  

3.4 互换机范例：

在之前的案例中，都没有互换机，生产者直接发送消息到队列。而一旦引入互换机，消息发送的模式会有很大变化：

可以看到，在订阅模型中，多了一个exchange角色，而且过程略有变化：

• Publisher：生产者，不再发送消息到队列中，而是发给互换机。
  
• Exchange：互换机，一方面，汲取生产者发送的消息。另一方面，知道如何处理惩罚消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的范例。
  
• Queue：消息队列也与以前一样，汲取消息、缓存消息。不过队列一定要与互换机绑定。
  
• Consumer：消费者，与以前一样，订阅队列，没有变化。
  

Exchange（互换机）只负责转发消息，不具备存储消息的本领，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！
互换机的范例有四种：

• Fanout：广播，将消息交给所有绑定到互换机的队列。
  
• Direct：订阅，基于 RoutingKey（路由 key）发送给订阅了消息的队列。
  
• Topic：通配符订阅，与 Direct 类似，只不过 RoutingKey 可以使用通配符。
  
• Headers：头匹配，基于 MQ 的消息头匹配，用的较少。
  

我们这里讲解前面三种互换机。
3.4.1 Fanout 互换机：

在广播模式下，消息发送流程是这样的：

总结：
互换机的特点：

• 汲取 publisher 发送的消息，将消息按照规则路由到与之绑定的队列。
  
• 不能缓存消息，路由失败，消息丢失。
  
• FanoutExchange 的会将消息路由到每个绑定的队列。
  

3.4.2 Direct 互换机：

在 Fanout 模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们盼望不同的消息被不同的队列消费。这时就要用到 Direct 范例的 Exchange。

在 Direct 模型下：

• 队列与互换机的绑定，不能是恣意绑定了，而是要指定一个RoutingKey（路由key）。
  
• 消息的发送方在向 Exchange 发送消息时，也必须指定消息的 RoutingKey。
  
• Exchange 不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判定，只有队列的Routingkey与消息的 Routing key完全一致，才会汲取到消息。
  

总结：
描述下 Direct 互换机与 Fanout 互换机的差异？

• Fanout 互换机将消息路由给每一个与之绑定的队列。
  
• Direct 互换机根据 RoutingKey 判定路由给哪个队列。
  
• 如果 Direct 互换机绑定的多个队列具有相同的 RoutingKey，则与 Fanout 功能类似。
  

3.4.3 Topic 互换机：

Topic范例的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。
只不过Topic范例Exchange可以让队列在绑定BindingKey 的时候使用通配符！
**BindingKey 一样寻常都是由一个或多个单词构成，多个单词之间以.分割。**例如： item.insert
通配符规则：

• #：匹配 0 个或多个词。
  
• *：匹配不多不少恰好 1 个词。
  

总结：
描述下 Direct 互换机与 Topic 互换机的差异？

• Topic 互换机与队列绑定时的 bindingKey 可以指定通配符。
  
• #：代表 0 个或多个词。
  
• *：代表 1 个词。
  

3.5 声明互换机和队列：

在之前我们都是基于RabbitMQ控制台来创建队列、互换机。但是在现实开发时，队列和互换机是步伐员定义的，未来项目上线，又要交给运维去创建。那么步伐员就需要把步伐中运行的所有队列和互换机都写下来，交给运维。在这个过程中是很轻易出现错误的。
因此推荐的做法是由步伐启动时查抄队列和互换机是否存在，如果不存在主动创建。
**声明互换机和队列一样寻常是消息的汲取者来做的。**这里的背景是微服务，消息的发送者和汲取者一样寻常是在不同的项目内里。
注意：如果项目中没有消费者（使用 @RabbitListener），可能创建互换机和队列会失败（MQ 上显示不出来，但是 spring 中对象成功创建），至于为什么会这样，我也不清晰，去网上找，问 AI 都没有这方面的资料。这个标题是我在项目中遇到的，解决了好久，算是比较小众的标题，如果有相怜悯况的友友，可以试试）。
3.5.1 使用 API：

SpringAMQP 提供了一个 Queue 类，用来创建队列：

SpringAMQP 还提供了一个 Exchange 接口，来表现所有不同范例的互换机：

我们可以自己创建队列和互换机，不过 SpringAMQP 还提供了 ExchangeBuilder 来简化这个过程：

而在绑定队列和互换机时，则需要使用 BindingBuilder 来创建 Binding 对象：

3.5.1.1 fanout 示例：

1. package com.itheima.consumer.config;
3. import org.springframework.amqp.core.Binding;
4. import org.springframework.amqp.core.BindingBuilder;
5. import org.springframework.amqp.core.FanoutExchange;
6. import org.springframework.amqp.core.Queue;
7. import org.springframework.context.annotation.Bean;
8. import org.springframework.context.annotation.Configuration;
10. @Configuration
11. public class FanoutConfig {
12.     /**
13.      * 声明交换机
14.      * @return Fanout类型交换机
15.      */
16.     @Bean
17.     public FanoutExchange fanoutExchange(){
18.         return new FanoutExchange("hmall.fanout");
19.     }
21.     /**
22.      * 第1个队列
23.      */
24.     @Bean
25.     public Queue fanoutQueue1(){
26.         return new Queue("fanout.queue1");
27.     }
29.     /**
30.      * 绑定队列和交换机
31.      */
32.     @Bean
33.     public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
34.         return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
35.     }
37.     /**
38.      * 第2个队列
39.      */
40.     @Bean
41.     public Queue fanoutQueue2(){
42.         return new Queue("fanout.queue2");
43.     }
45.     /**
46.      * 绑定队列和交换机
47.      */
48.     @Bean
49.     public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
50.         return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
51.     }
52. }

复制代码

3.5.1.2 direct 示例：

direct 模式由于要绑定多个 KEY，会非常贫苦，每一个 Key 都要编写一个 binding：

1. package com.itheima.consumer.config;
3. import org.springframework.amqp.core.*;
4. import org.springframework.context.annotation.Bean;
5. import org.springframework.context.annotation.Configuration;
7. @Configuration
8. public class DirectConfig {
10.     /**
11.      * 声明交换机
12.      * @return Direct类型交换机
13.      */
14.     @Bean
15.     public DirectExchange directExchange(){
16.         return ExchangeBuilder.directExchange("hmall.direct").build();
17.     }
19.     /**
20.      * 第1个队列
21.      */
22.     @Bean
23.     public Queue directQueue1(){
24.         return new Queue("direct.queue1");
25.     }
27.     /**
28.      * 绑定队列和交换机
29.      */
30.     @Bean
31.     public Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){
32.         return BindingBuilder.bind(directQueue1).to(directExchange).with("red");
33.     }
34.     /**
35.      * 绑定队列和交换机
36.      */
37.     @Bean
38.     public Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){
39.         return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");
40.     }
42.     /**
43.      * 第2个队列
44.      */
45.     @Bean
46.     public Queue directQueue2(){
47.         return new Queue("direct.queue2");
48.     }
50.     /**
51.      * 绑定队列和交换机
52.      */
53.     @Bean
54.     public Binding bindingQueue2WithRed(Queue directQueue2, DirectExchange directExchange){
55.         return BindingBuilder.bind(directQueue2).to(directExchange).with("red");
56.     }
57.     /**
58.      * 绑定队列和交换机
59.      */
60.     @Bean
61.     public Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange){
62.         return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");
63.     }
64. }

复制代码

由于 Topic 互换机的声明和 Direct 差不多，大家照着上面的 Direct ，修改一下范例就能成功创建，这里就不演示了。
3.5.2 使用注解：

基于 @Bean 的方式声明队列和互换机比较贫苦，尤其是 direct 互换机，Spring 还提供了基于注解方式来声明。
3.5.2.1 fanout 示例:

1. @RabbitListener(bindings = @QueueBinding(
2.         value = @Queue(name = "fanout.queue1"),
3.         exchange = @Exchange(name = "hmall.fanout", type = ExchangeTypes.FANOUT)
4. ))
5. public void listenFanoutQueue1(String msg){
6.     System.out.println("消费者1接收到fanout.queue1的消息：【" + msg + "】");
7. }
9. @RabbitListener(bindings = @QueueBinding(
10.         value = @Queue(name = "fanout.queue2"),
11.         exchange = @Exchange(name = "hmall.fanout", type = ExchangeTypes.FANOUT)
12. ))
13. public void listenFanoutQueue2(String msg){
14.     System.out.println("消费者2接收到fanout.queue2的消息：【" + msg + "】");
15. }

复制代码

3.5.2.2 direct 示例：

1. @RabbitListener(bindings = @QueueBinding(
2.     value = @Queue(name = "direct.queue1"),
3.     exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),
4.     key = {"red", "blue"}
5. ))
6. public void listenDirectQueue1(String msg){
7.     System.out.println("消费者1接收到direct.queue1的消息：【" + msg + "】");
8. }
10. @RabbitListener(bindings = @QueueBinding(
11.     value = @Queue(name = "direct.queue2"),
12.     exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),
13.     key = {"red", "yellow"}
14. ))
15. public void listenDirectQueue2(String msg){
16.     System.out.println("消费者2接收到direct.queue2的消息：【" + msg + "】");
17. }

复制代码

3.6 消息转化器：

convertAndSend 方法可以恣意范例的消息。

而在数据传输时，它会把你发送的消息序列化为字节发送给 MQ，汲取消息的时候，还会把字节反序列化为 Java 对象。
只不过，默认情况下 Spring 采用的序列化方式是 JDK 序列化。JDK 序列化存在下列标题：

• 数据体积过大
  
• 有安全弊端
  
• 可读性差
  

所以我们需要换一个消息转化器。
这里采用 JSON 方式来做序列化和反序列化。
在发送者和接受者两个服务中都引入依赖。

1. <dependency>
2.     <groupId>com.fasterxml.jackson.dataformat</groupId>
3.     <artifactId>jackson-dataformat-xml</artifactId>
4.     <version>2.9.10</version>
5. </dependency>

复制代码

注意，如果项目中引入了spring-boot-starter-web依赖，则无需再次引入Jackson依赖。
设置消息转换器，在发送者和接受者两个服务的启动类中添加一个 Bean 即可：

1. @Bean
2. public MessageConverter messageConverter(){
3.     // 1.定义消息转换器
4.     Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
5.     // 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
6.     jackson2JsonMessageConverter.setCreateMessageIds(true);
7.     return jackson2JsonMessageConverter;
8. }

复制代码

消息转换器中添加的 messageId 可以便于我们未来做幂等性判定。
参考文献：

• 黑马步伐员
  

结语：
实在写博客不但仅是为了教大家，同时这也有利于我巩固知识点，和做一个学习的总结，由于作者程度有限，对文章有任何标题还请指出，非常感谢。如果大家有所劳绩的话，还请不要吝啬你们的点赞收藏和关注，这可以激励我写出更加精良的文章。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。