《微服务实战》第十四章 RabbitMQ应用

分手后的思念是犯贱 2023-10-10 20:44 15阅读 0赞

## 系列文章目录 ##

[第十六章 Spring cloud stream应用][Spring cloud stream]  
[第十五章 RabbitMQ 延迟队列][RabbitMQ]  
[第十四章 RabbitMQ应用][RabbitMQ 1]

![在这里插入图片描述][28e7d15c967d4b979b3f0f65fb7c5e04.png]

--------------------

### 文章目录  系列文章目录  @\[TOC\](文章目录)  前言  1、RabbitMQ概念概念  1.1、生产者和消费者  1.2、队列  1.3、交换机、路由键、绑定  1.3.1、交换机类型  2、RabbitMQ运转流程  2.1、生产者发送消息流程  2.2、消费者接收消息的过程  2.3、AMQP协议  3、RabbitMQ windows安装  3.1、下载  3.2、安装  4、Spring Boot 整合RabbitMQ  4.1、在user-service添加依赖  4.2、配置文件添加  4.3、增加RabbitMQ配置类  4.4、新增消费监听类  4.5、消息生产端 ###

## 前言 ##

一般MQ用于系统解耦、削峰使用，常见于微服务、业务活动等场景。

## 1、RabbitMQ概念概念 ##

RabbitMQ整体上是一个生产者与消费者模型，主要负责接收、存储和转发消息。  
![在这里插入图片描述][e6bdf6406f2b46b397f2163b97077375.png]

### 1.1、生产者和消费者 ###

*  Producer：生产者，就是投递消息的一方。消息一般可以包含2个部分：消息体和标签(Label)。消息的标签用来描述这条消息，比如一个交换器的名称和一个路由键。
 *  Consumer：消费者，就是接受消息的一方。消费者连接到RabbitMQ服务器，并订阅到队列上。当消费者消费一条消息时，只是消费消息的消息体(payload)
 *  Broker：消息中间件的服务节点。一个RabbitMQ Broker看做一台RabbitMQ服务器  
    ![在这里插入图片描述][10d63065d58a41f5bf06a047d856718a.png]

### 1.2、队列 ###

Queue：队列，是RabbitMQ的内部对象，用于存储消息  
![在这里插入图片描述][299a84ecd20d47f0a7552d0e7f2fe407.png]  
![在这里插入图片描述][f9400566201747ad9eb2c78bb3e6852d.png]

### 1.3、交换机、路由键、绑定 ###

Exchange：交换器。生产者将消息发送到Exchange(交换器，通常也可以用大写的"X"来表示)，有交换器将消息路由到一个或者多个队列中。如果路由不到，或许会返回给生产者，或许直接丢弃。  
![在这里插入图片描述][33567a68679f4d1493eb431e60ed4727.png]  
RoutingKey：路由键。生产者将消息发给交换器的时候，一般会指定一个RoutingKey，用来指定这个消息的路由规则，而这个RoutingKey需要与交换器类型和绑定键(BindingKey)联合使用才能最终生效。  
Binding：绑定。RabbitMQ中通过绑定将交换器与队列联合起来，在绑定的时候一般会指定一个绑定键(BindingKey)，这样RabbitMQ就知道如何正确地将消息路由到队列了。  
![在这里插入图片描述][f66a9f86890842de803f30f2b2da293b.png]

#### 1.3.1、交换机类型 ####

*  Direct Exchange：直连交换机，根据Routing Key(路由键)进行投递到不同队列。  
    ![在这里插入图片描述][b6e33f0a745b493e90b0a37a85b0629f.png]  
    ![在这里插入图片描述][cc2543f7815a4da49b9bcb09bb6e24e4.png]
 *  Fanout Exchange：扇形交换机，采用广播模式，根据绑定的交换机，路由到与之对应的所有队列。  
    ![在这里插入图片描述][a386c82b9cb8436c9996df65133932e7.png]
 *  Topic Exchange：主题交换机，对路由键进行模式匹配后进行投递，符号\#表示一个或多个词，\*表示一个词。  
    ![在这里插入图片描述][23d973a1fea84d0fae1e293859a492d5.png]
 *  Header Exchange：头交换机，不处理路由键。而是根据发送的消息内容中的headers属性进行匹配。  
    ![在这里插入图片描述][be435a74d5d644fba5fa55d2cd491176.png]  
    自学参考：https://blog.csdn.net/qq\_38550836/article/details/95358353

## 2、RabbitMQ运转流程 ##

### 2.1、生产者发送消息流程 ###

*  生产者连接到RabbitMQ Broker，建立一个连接(Connection)，开启一个信道(Channel)
 *  生产者声明一个交换器，并设置相关属性，比如交换器类型、是否持久化等
 *  生产者声明一个队列并设置相关属性，比如是否排他、是否持久化、是否自动删除等
 *  生产者通过路由键将交换器和队列绑定起来
 *  生产者发送消息至RabbitMQ Broker，其中包含路由键、交换器等信息
 *  相应的交换器根据接收到的路由键查找相匹配的队列。
 *  如果找到，则将从生产者发送过来的消息存入相应的队列。
 *  如果没有找到，则根据生产者配置的属性选择丢弃还是回退给生产者
 *  关闭信道
 *  关闭连接

### 2.2、消费者接收消息的过程 ###

*  消费者连接到RabbitMQ Broker，建立一个连接(Connection)，开启一个信道(Channel)。
 *  消费者向RabbitMQ Broker请求消费相应队列中的消息，可能会设置相应的回调函数，以及做一些准备工作。
 *  等待RabbitMQ Broker回应并投递相应队列中队列的消息，消费者接收消息。
 *  消费者确认(ack)接收到的消息。
 *  RabbitMQ从队列中删除相应已经被确认的消息。
 *  关闭信道
 *  关闭连接  
    ![在这里插入图片描述][b59157029cf5409aa52e16acd3ac7dbd.png]无论是生产者还是消费者，都需要和RabbitMQ Broker建立连接，这个连接就是一条TCP连接，也就是Connection。一旦TCP连接建立起来，客户端紧接着可以创建一个AMQP信道(Channel)，每个信道都会被指派一个唯一的ID。信道是建立在Connection之上的虚拟连接，RabbitMQ处理的每条AMQP指令都是通过信道完成的。

### 2.3、AMQP协议 ###

Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有RabbitMQ等  
![在这里插入图片描述][e047a790b4144ae6ba08259e4f144918.png]

*  Broker：接收和分发消息的应用，RabbitMQ 就是 Message Broker
 *  Virtual Host：虚拟 Broker，将多个单元隔离开
 *  Connection：publisher / consumer 和 broker 之间的 tcp 连接
 *  Channel：connection 内部建立的逻辑连接，通常每个线程创建单独的 channel
 *  Routing key：路由键，用来指示消息的路由转发，相当于快递的地址
 *  Exchange：交换机，相当于快递的分拨中心
 *  Queue：消息队列，消息最终被送到这里等待 consumer 取走
 *  Binding：exchange 和 queue 之间的虚拟连接，用于 message 的分发依据

## 3、RabbitMQ windows安装 ##

### 3.1、下载 ###

https://github.com/erlang/otp/releases/download/OTP-25.2/otp\_win64\_25.2.exe  
https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.11.5/rabbitmq-server-3.11.5.exe

### 3.2、安装 ###

配置环境变量  
![在这里插入图片描述][8ecc9853ed5d4026893f92635819ba57.png]  
![在这里插入图片描述][1ef6cba171e348fd82d209904900a89b.png]

> cd D:\\Program Files\\RabbitMQ Server\\rabbitmq\_server-3.11.5\\sbin

开启rabbitmq-plugins插件

> rabbitmq-plugins enable rabbitmq\_management  
> ![在这里插入图片描述][a62d263376464ad6b79078179690d8cc.png]

打开地址  
http://127.0.0.1:15672/  
![在这里插入图片描述][b435855970954417ad3c2d2217102de2.png]  
输入用户名/密码：guest/guest

## 4、Spring Boot 整合RabbitMQ ##

### 4.1、在user-service添加依赖 ###

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

### 4.2、配置文件添加 ###

spring:
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest
        password: guest

### 4.3、增加RabbitMQ配置类 ###

package com.xxxx.user.config;
    
    import org.springframework.amqp.core.*;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class RabbitMQConfig {
        
        /******************direct**********************/
        /**
         * 创建direct队列
         * @return
         */
        @Bean
        public Queue directQueue(){
        
            return new Queue("directQueue");
        }
    
        /**
         * 创建direct交换机
         * @return
         */
        @Bean
        public DirectExchange directExchange(){
        
            return new DirectExchange("directExchange");
        }
    
        /**
         * 把队列和交换机绑定在一起
         * @param queue
         * @param directExchange
         * @return
         */
        @Bean
        public Binding bindingDirect(@Qualifier("directQueue") Queue queue, DirectExchange directExchange){
        
            return BindingBuilder.bind(queue).to(directExchange).with("routingKey");
        }
    
        /******************topic**********************/
        @Bean
        public Queue topicQuerue1(){
        
            return new Queue("topicQuerue1");
        }
    
        @Bean
        public Queue topicQuerue2(){
        
            return new Queue("topicQuerue2");
        }
        @Bean
        public TopicExchange topicExchange(){
        
            return new TopicExchange("topicExchange");
        }
        @Bean
        public Binding bindingTopic1(@Qualifier("topicQuerue1") Queue queue,@Qualifier("topicExchange") TopicExchange topicExchange){
        
            return BindingBuilder.bind(queue).to(topicExchange).with("topic.key1");
        }
    
        /**
         * 通配符：* 表示一个词，# 表示零个或多个词
         * @param queue
         * @param topicExchange
         * @return
         */
        @Bean
        public Binding bindingTopic2(@Qualifier("topicQuerue2") Queue queue,@Qualifier("topicExchange") TopicExchange topicExchange){
        
            return BindingBuilder.bind(queue).to(topicExchange).with("topic.#");
        }
    
        /******************fanout**********************/
        @Bean
        public Queue fanoutQueue1(){
        
            return new Queue("fanoutQueue1");
        }
        @Bean
        public Queue fanoutQueue2(){
        
            return new Queue("fanoutQueue2");
        }
        @Bean
        public Queue fanoutQueue3(){
        
            return new Queue("fanoutQueue3");
        }
        @Bean
        public FanoutExchange fanoutExchange(){
        
            return new FanoutExchange("fanoutExchange");
        }
    
        @Bean
        public Binding bindingFanout1(@Qualifier("fanoutQueue1") Queue queue,@Qualifier("fanoutExchange") FanoutExchange fanoutExchange){
        
            return BindingBuilder.bind(queue).to(fanoutExchange);
        }
    
        @Bean
        public Binding bindingFanout2(@Qualifier("fanoutQueue2") Queue queue,@Qualifier("fanoutExchange") FanoutExchange fanoutExchange){
        
            return BindingBuilder.bind(queue).to(fanoutExchange);
        }
    
        @Bean
        public Binding bindingFanout3(@Qualifier("fanoutQueue3") Queue queue,@Qualifier("fanoutExchange") FanoutExchange fanoutExchange){
        
            return BindingBuilder.bind(queue).to(fanoutExchange);
        }
    }

### 4.4、新增消费监听类 ###

package com.xxxx.user.consumer;
    
    import com.hqyj.drp.common.entity.UserInfo;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    @Slf4j
    @RabbitListener(queues = "directQueue")
    public class DataDirectReceiver {
        
        @RabbitHandler
        public void process(String data){
        
            log.info("收到directQueue队列信息：" + data);
        }
    
        @RabbitHandler
        public void process(UserInfo data){
        
            log.info("收到directQueue队列信息：" + data);
        }
    }

package com.xxxx.user.consumer;
    
    import com.xxxx.common.entity.UserInfo;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    @Slf4j
    @RabbitListener(queues = {
        "topicQuerue1","topicQuerue2"})
    public class DataFanoutReceiver {
        
        @RabbitHandler
        public void process(String data){
        
            log.info("收到topicQuerue队列信息：" + data);
        }
    
        @RabbitHandler
        public void process(UserInfo data){
        
            log.info("收到topicQuerue队列信息：" + data);
        }
    }

package com.xxxx.user.consumer;
    
    import com.xxxx.common.entity.UserInfo;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    @Slf4j
    @RabbitListener(queues = {
        "fanoutQueue1","fanoutQueue2","fanoutQueue3"})
    public class DataTopicReceiver {
        
        @RabbitHandler
        public void process(String data){
        
            log.info("收到topicQuerue队列信息：" + data);
        }
    
        @RabbitHandler
        public void process(UserInfo data){
        
            log.info("收到topicQuerue队列信息：" + data);
        }
    }

### 4.5、消息生产端 ###

package com.xxxx.user;
    
    import com.xxxx.common.entity.UserInfo;
    import org.junit.jupiter.api.Test;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    @SpringBootTest
    public class DataSender {
        
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Test
        public void sendDirect(){
        
            UserInfo userInfo = new UserInfo();
            userInfo.setUserAccount("tiger");
            userInfo.setPassword("12345");
            this.rabbitTemplate.convertAndSend("directExchange","routingKey",userInfo);
        }
        @Test
        public void sendTopic(){
        
            this.rabbitTemplate.convertAndSend("topicExchange","topic.key2","Hello world topic");
        }
    
        @Test
        public void sendFanout(){
        
            this.rabbitTemplate.convertAndSend("fanoutExchange","","Hello world topic");
        }
    }

[Spring cloud stream]: https://blog.csdn.net/s445320/article/details/130826012
[RabbitMQ]: https://blog.csdn.net/s445320/article/details/130823776
[RabbitMQ 1]: https://blog.csdn.net/s445320/article/details/130822527
[28e7d15c967d4b979b3f0f65fb7c5e04.png]: https://img-blog.csdnimg.cn/28e7d15c967d4b979b3f0f65fb7c5e04.png
[e6bdf6406f2b46b397f2163b97077375.png]: https://img-blog.csdnimg.cn/e6bdf6406f2b46b397f2163b97077375.png
[10d63065d58a41f5bf06a047d856718a.png]: https://img-blog.csdnimg.cn/10d63065d58a41f5bf06a047d856718a.png
[299a84ecd20d47f0a7552d0e7f2fe407.png]: https://img-blog.csdnimg.cn/299a84ecd20d47f0a7552d0e7f2fe407.png
[f9400566201747ad9eb2c78bb3e6852d.png]: https://img-blog.csdnimg.cn/f9400566201747ad9eb2c78bb3e6852d.png
[33567a68679f4d1493eb431e60ed4727.png]: https://img-blog.csdnimg.cn/33567a68679f4d1493eb431e60ed4727.png
[f66a9f86890842de803f30f2b2da293b.png]: https://img-blog.csdnimg.cn/f66a9f86890842de803f30f2b2da293b.png
[b6e33f0a745b493e90b0a37a85b0629f.png]: https://img-blog.csdnimg.cn/b6e33f0a745b493e90b0a37a85b0629f.png
[cc2543f7815a4da49b9bcb09bb6e24e4.png]: https://img-blog.csdnimg.cn/cc2543f7815a4da49b9bcb09bb6e24e4.png
[a386c82b9cb8436c9996df65133932e7.png]: https://img-blog.csdnimg.cn/a386c82b9cb8436c9996df65133932e7.png
[23d973a1fea84d0fae1e293859a492d5.png]: https://img-blog.csdnimg.cn/23d973a1fea84d0fae1e293859a492d5.png
[be435a74d5d644fba5fa55d2cd491176.png]: https://img-blog.csdnimg.cn/be435a74d5d644fba5fa55d2cd491176.png
[b59157029cf5409aa52e16acd3ac7dbd.png]: https://img-blog.csdnimg.cn/b59157029cf5409aa52e16acd3ac7dbd.png
[e047a790b4144ae6ba08259e4f144918.png]: https://img-blog.csdnimg.cn/e047a790b4144ae6ba08259e4f144918.png
[8ecc9853ed5d4026893f92635819ba57.png]: https://img-blog.csdnimg.cn/8ecc9853ed5d4026893f92635819ba57.png
[1ef6cba171e348fd82d209904900a89b.png]: https://img-blog.csdnimg.cn/1ef6cba171e348fd82d209904900a89b.png
[a62d263376464ad6b79078179690d8cc.png]: https://img-blog.csdnimg.cn/a62d263376464ad6b79078179690d8cc.png
[b435855970954417ad3c2d2217102de2.png]: https://img-blog.csdnimg.cn/b435855970954417ad3c2d2217102de2.png