SpringCloud学习笔记（八）消息中间件RabbitMQ

ゝ一世哀愁。 2022-03-19 15:44 501阅读 0赞

**目录**

一、简介

二、 windows环境下的安装

（1）下载并安装 Eralng

（2）下载并安装rabbitmq

（3）安装管理界面（插件）

（4）重新启动服务（5）登录

三、docker环境下的安装

1.镜像下载

2.创建容器

四、rabbitmq三种模式

（a）直接模式 (direct)

（b）分裂模式 (fanout)

（c）主题模式（topic）

五 springboot2.0之后，rabbitmq的简单用法

1、通过手动创建myQueue

2、自动生成消息队列（@Since 2.0）

3、将消息队列和交换器进行绑定

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

## 一、简介 ##

消息队列中间件是分布式系统中重要的组件，主要解决应用耦合，异步消息，流量 削锋等问题实现高性能，高可用，可伸缩和终一致性\[架构\]   使用较多的消息队列有 ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ  
以下介绍消息队列在实际应用中常用的使用场景：**异步处理**，**应用解耦**，**流量削锋**和**消息通讯**四个场景

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70][]

**RabbitMQ Server**： 也叫broker server，它是一种**传输服务**。 角色就是维护一条 从Producer到Consumer的路线，保证数据能够按照指定的方式进行传输。  
**Producer**： 消息生产者，如图A、B、C，数据的发送方。消息生产者连接RabbitMQ服 务器然后将消息投递到Exchange。  
**Consumer**：消息消费者，如图1、2、3，数据的接收方。消息消费者订阅队列， RabbitMQ将Queue中的消息发送到消息消费者。 Exchange：生产者将消息发送到Exchange（交换器），由Exchange将消息路由到一个 或多个Queue中（或者丢弃）。Exchange并不存储消息。*RabbitMQ中的Exchange有 direct、fanout、topic、headers四种类型*，每种类型对应不同的路由规则。  
**Queue**：（队列）是RabbitMQ的内部对象，用于存储消息。消息消费者就是通过订阅 队列来获取消息的，RabbitMQ中的消息都只能存储在Queue中，生产者生产消息并终 投递到Queue中，消费者可以从Queue中获取消息并消费。多个消费者可以订阅同一个 Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者 都收到所有的消息并处理。  
**RoutingKey**：生产者在将消息发送给Exchange的时候，一般会指定一个routing key， 来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联 合使用才能终生效。在Exchange Type与binding key固定的情况下（在正常使用时一 般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过 指定routing key来决定消息流向哪里。RabbitMQ为routing key设定的长度限制为255 bytes。

## 二、 windows环境下的安装  ##

###    （1）下载并安装 Eralng ###

配套软件中已提供otp\_win64\_20.2.exe   （以管理员身份运行安装）

### （2）下载并安装rabbitmq ###

配套软件中已提供rabbitmq-server-3.7.4.exe。双击安装，**注意不要安装在包含中文和 空格的目录下**！安装后window服务中就存在rabbitMQ了，并且是启动状态。

###    （3）安装管理界面（插件） ###

进入rabbitMQ安装目录的sbin目录，输入命令

> rabbitmq‐plugins enable rabbitmq\_management

### （4）重新启动服务   （5）登录 ###

打开浏览器，地址栏输入http://127.0.0.1:15672  ,即可看到管理界面的登陆页

## **三、docker环境下的安装 ** ##

### 1.镜像下载 ###

> docker pull rabbitmq:management

### 2.创建容器 ###

rabbitmq需要有映射以下端口:  5671   5672  4369  15671   15672   25672

15672 (if management plugin is enabled)

15671  management监听端口

5672, 5671 (AMQP 0-9-1 without and with TLS)

4369 (epmd)   epmd  代表 Erlang 端口映射守护进程 25672 (Erlang distribution)

> docker run -di --name=tensquare\_rabbitmq -p 5671:5617 -p 5672:5672 -p 4369:4369 -p 15671:15671 -p 15672:15672 -p 25672:25672 rabbitmq:management

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 1][]

## 四、rabbitmq三种模式 ##

直接模式（Direct）、分裂模式(Fanout)、主题模式(Topic)

### （a）直接模式 (direct) ###

走默认交换器（“”），类似不走交换器，直接将消息发送给队列

使用场景：将消息发送给一个节点时使用

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 2][]

***任何发送到Direct Exchange的消息都会被转发到RouteKey中指定的Queue。***  
  *1.一般情况可以使用rabbitMQ自带的Exchange：”"(该Exchange的名字为空字符串，下 文称其为default Exchange)*。

2.这种模式下不需要将Exchange进行任何绑定(binding)操作

3.消息传递时需要一个“RouteKey”，可以简单的理解为要发送到的队列名字。

4.如果vhost中不存在RouteKey中指定的队列名，则该消息会被抛弃。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 3][]

新建测试类：**生产者测试类**

@RunWith(SpringRunner.class)
    @SpringBootTest(classes = RabbitApplication.class)
    public class ProductTest {
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Test
        public void testSend(){
            rabbitTemplate.convertAndSend("itcast","直接模式测试");
        }
    }

发送消息后，消息队列会有一条消息等待被消费

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 4][]

新建消费者类:  queues 指明消费的队列

>     @Component
>     @RabbitListener(queues = "itcast")
>     public class Customer1 {
>     
>         @RabbitHandler
>         public void getMsg(String msg){
>             System.out.println("直接模式消费信息：" + msg);
>         }
>     }

消息内部做的有负载均衡，会依次将消息进行分发

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 5][]

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

### （b）分裂模式 (fanout) ###

**将消息发送给多个队列时使用**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 6][]

步骤一：新建三个队列queue

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 7][]

步骤二：新建交换机：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 8][]

步骤三：实现交换机和队列的绑定

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 9][]

步骤四：

新建分裂模式的消费者

>     /**
>     * 分裂者模式测试
>     */
>     @Test
>     public void testFanout(){
>         rabbitTemplate.convertAndSend("chuanzhi","","分裂者模式分发消息");
>     }

新建交换器对应的消费者1

>     @Component
>     @RabbitListener(queues = "itcast")
>     public class Customer1 {
>     
>         @RabbitHandler
>         public void getMsg(String msg){
>             System.out.println("itcast消费信息：" + msg);
>         }
>     }

新建交换器对应的消费者2

>     @Component
>     @RabbitListener(queues = "itheima")
>     public class Customer2 {
>     
>         @RabbitHandler
>         public void getMsg(String msg){
>             System.out.println("itheima消费信息：" + msg);
>         }
>     }

新建交换器对应的消费者3

>     @Component
>     @RabbitListener(queues = "kudingyu")
>     public class Customer3 {
>     
>         @RabbitHandler
>         public void getMsg(String msg){
>             System.out.println("kudingyu消费信息：" + msg);
>         }
>     }

运行测试类和启动类

![20190217233345263.png][]

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

### （c）主题模式（topic） ###

原理

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 10][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 11][]

绑定队列

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 12][]

> 方式一：只往itcast队列发送消息  
> rabbitTemplate.convertAndSend("beijing","good.abc.123","主题模式分发消息");

>  方式二：只往itheima发送
> 
> rabbitTemplate.convertAndSend("beijing","abc.log","主题模式分发消息");

> 方式三：三种方式都发送
> 
> rabbitTemplate.convertAndSend("beijing","good.log","主题模式分发消息");

实际应用：使用map进行接收，如果发送形式不同，会陷入死循环

> 生产者 
> 
>     Map<String,String> map = new HashMap<>();
>     map.put("mobile",mobile);
>     map.put("checkCode",checkCode);
>     rabbitTemplate.convertAndSend("sms",map);

> 消费者
> 
>     @Component
>     @RabbitListener(queues = "sms")
>     public class SmsListener {
>     
>         @RabbitHandler
>         public void executeSms(Map<String,String> map){
>             System.out.println("手机号码：" + map.get("mobile"));
>             System.out.println("短信验证码：" + map.get("checkCode"));
>         }
>     }

注意：

rabbitmq的默认账号、密码：guest/guest

但是如果需要远程连接，需要新建用户

## 五 springboot2.0之后，rabbitmq的简单用法 ##

设置消息监听

### 1、通过手动创建myQueue ###

import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.annotation.Exchange;
    import org.springframework.amqp.rabbit.annotation.Queue;
    import org.springframework.amqp.rabbit.annotation.QueueBinding;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    /**
     * 接收mq消息
     */
    @Component
    @Slf4j
    public class MqReceiver {
    
        //  1、方式1，通过手动新建myQueue 
        @RabbitListener(queues = "myQueue")
        public void process (String message) {
    
              log.info("MqReceiver:{}",message);
        }
    }

### 2、自动生成消息队列（@Since 2.0） ###

import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.annotation.Exchange;
    import org.springframework.amqp.rabbit.annotation.Queue;
    import org.springframework.amqp.rabbit.annotation.QueueBinding;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    /**
     * 接收mq消息
     */
    @Component
    @Slf4j
    public class MqReceiver {
    
       
        //  2、自动生成 @Since 2.0,本项目不能使用
        @RabbitListener(queuesToDeclare = @Queue("myAutomitic"))
        public void process (String message) {
    
              log.info("MqReceiver:{}",message);
        }
    }

### 3、将消息队列和交换器进行绑定 ###

import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.annotation.Exchange;
    import org.springframework.amqp.rabbit.annotation.Queue;
    import org.springframework.amqp.rabbit.annotation.QueueBinding;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    /**
     * 接收mq消息
     */
    @Component
    @Slf4j
    public class MqReceiver {
    
      
        //  3、进行绑定操作
        @RabbitListener(bindings = @QueueBinding(
                value = @Queue("myQueue"),
                exchange = @Exchange("myExchange")
        ))
        public void process (String message) {
    
              log.info("MqReceiver:{}",message);
        }
    }

发送端使用的是测试代码

@Test
        public void sendMqMsg() {
            amqpTemplate.convertAndSend("myQueue","now:" + new Date());
        }

六 订单和产品服务相互发送消息

1、产品服务扣库存

>     amqpTemplate.convertAndSend("productInfo", JsonUtil.toJson(productInfoOutput));

2、订单服务接收消息

>     @RabbitListener(queuesToDeclare = @Queue("productInfo"))
>     public void productProcess (String message) {
>         // message => ProductInfoOutput
>     
>         ProductInfoOutput productInfoOutput = (ProductInfoOutput) JsonUtil.fromJson(message, ProductInfoOutput.class);
>         log.info("消息队列【{}】,消息内容是:{}","productInfo", productInfoOutput);
>     }

3、jsonUtil

public class JsonUtil {
    
    	private static ObjectMapper objectMapper = new ObjectMapper();
    
    	/**
    	 * 转换为json字符串
    	 * @param object
    	 * @return
    	 */
    	public static String toJson(Object object) {
    		try {
    			return objectMapper.writeValueAsString(object);
    		} catch (JsonProcessingException e) {
    			e.printStackTrace();
    		}
    		return null;
    	}
    
    	/**
    	 * 将字符串转为实体
    	 * @param message
    	 * @param object
    	 * @return
    	 */
    	public static Object fromJson(String message,Class object){
    		try {
    			return objectMapper.readValue(message, object);
    		} catch (JsonProcessingException e){
    			e.printStackTrace();
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    		return null;
    	}
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70]: /images/20220319/f01d19fd1ad04c129f517250a7b8b494.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 1]: /images/20220319/a0c0b27bca824255a55df892d7d50c8d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 2]: /images/20220319/81520bc79e2140cbaba9c30881a2601f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 3]: /images/20220319/56bc18592d054665af0c3e465a08adc9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 4]: /images/20220319/bc923514ba3c4769b75e273d8d5cf1b6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 5]: /images/20220319/5996f6fca43c4ccbb0c005a0603a100b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 6]: /images/20220319/532cae1956f44b5ea12a9357e8a5621a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 7]: /images/20220319/3264e733c7bd46198c6332b2b7fab908.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 8]: /images/20220319/cdcf9f5cce1d442490b08cc52cc9a65f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 9]: /images/20220319/12487f2734fc492496f3e6d5e572a4df.png
[20190217233345263.png]: /images/20220319/e968e829bee44409bf78d14d6283e57b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 10]: /images/20220319/694a17a6f1d942aa82dccbb4fcfc5c56.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 11]: /images/20220319/1534b0f2c3e745fa8432d884e9834a59.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70 12]: /images/20220319/3c4441f17efa4527949d7b4c268e14d1.png