MQ消息队列的基础利用（以RabbitMQ为例）

一、同步调用与异步调用

1、同步调用

同步调用长处：
时效性强，可以立即得到结果
同步调用存在的题目
（1）耦合度高：每次加入新的需求，都要修改原来的代码
（2）性能下降：调用者必要期待服务提供者相应，如果调用链过长则相应时间即是每次调用的时间之和
（3）资源浪费：调用链中的每个服务在期待相应过长中，不能开释请求占用的资源，高并发场景下会极度浪费体系资源
（4）级联失败：如果服务提供者出现题目，所有调用方都会跟着出题目，犹如多米诺骨牌一样，迅速导致整个微服务群故障
2、异步调用

长处：

• 耦合度低
  
• 吞吐量提高
  
• 故障隔离
  
• 流量削峰
  

缺点：

• 依赖于Broker的可靠性、安全性、吞吐能力
  
• 架构复杂，业务没有明显的流程线，欠好追踪管理
  

二、初识MQ

MQ（MessageQueue），中文是消息队列，字面意思就是存放消息的队列。也就是变乱驱动架构中的Broker。

三、RabbitMQ

docker安装RabbitMQ步调
1、拉取镜像

1. docker pull rabbitmq

复制代码

2、安装RabbitMQ，实行下令

1. docker run -d -p 15672:15672 -p 5672:5672 \
2.         -e RABBITMQ_DEFAULT_USER=admin \
3.         -e RABBITMQ_DEFAULT_PASS=Miss5016 \
4.         --hostname mq1 \
5.         --name my-rabbitmq\
6.         rabbitmq:latest

复制代码

3、开放防火墙和安全组对应的端口

1. # 开放端口防火墙
2. firewall-cmd --permanent --zone=public --add-port=15672/tcp
3. firewall-cmd --permanent --zone=public --add-port=5672/tcp
4. # 重启防火墙
5. systemctl restart firewalld.service

复制代码

4、启用RabbitMQ的web端管理界面

1. # 进入容器内部
2. docker exec -it my-rabbitmq /bin/bash
4. # 允许使用web端管理
5. rabbitmq-plugins enable rabbitmq_management

复制代码

5、通过浏览器访问，ip+端标语

6、输入docker run下令时的账号暗码即可进入访问

7、题目
通道被禁用 Stats in management UI are disabled on this node

解决方法：
（1）进入容器内部

1. docker exec -it my-rabbitmq /bin/bash

复制代码

（2）切换到RabbitMQ的设置目录下

1. cd /etc/rabbitmq/conf.d/

复制代码

（3）修改设置

1. echo management_agent.disable_metrics_collector = false > management_agent.disable_metrics_collector.conf

复制代码

（4）退出并重启docker容器

1. # 从容器内部退出
2. exit
3. # 重启容器
4. docker restart my-rabbitmq

复制代码

（5）重新访问通道

RabbitMQ的结构和概念

四、RabbitMQ的消息模子

官网文档：https://www.rabbitmq.com/tutorials
1、根本消息队列（BasicQueue）

总共分为三个角色：

• publisher：消息发布者，将消息发送到队列queue
  
• queue：消息队列，负责接受消息并缓存消息
  
• consumer：订阅队列，处置惩罚队列中的消息
  

示例，通过Springboot的测试类实现
（1）导入依赖

1. <dependency>
2.     <groupId>com.rabbitmq</groupId>
3.     <artifactId>amqp-client</artifactId>
4.     <version>5.20.0</version>
5. </dependency>

复制代码

（2）生产者代码

1. @Test
2.     void contextLoads() {
3.         try {
4.             //1、建立连接
5.             ConnectionFactory factory = new ConnectionFactory();
6.             //2、设置连接参数
7.             factory.setHost("ip地址"); //主机名
8.             factory.setPort(5672); //连接端口
9.             factory.setVirtualHost("/"); //虚拟主机
10.             factory.setUsername("用户名"); //用户名
11.             factory.setPassword("密码"); //密码
12.             //3、建立连接
13.             Connection connection = factory.newConnection();
14.             //4、创建通道
15.             Channel channel = connection.createChannel();
16.             //5、创建队列并声明队列
17.             String queueName = "simple.queue"; //队列名称
18.             channel.queueDeclare(queueName,false, false, false, null);
19.             //6、发送消息
20.             String message = "hello rabbitMQ";
21.             channel.basicPublish("", queueName, null, message.getBytes());
22.             System.out.println("消息发送完毕。。。");
23.             //关闭管道和连接
24.             channel.close();
25.             connection.close();
26.         }catch (Exception e) {
27.             System.out.println("生产者发送消息失败 ----- > " + e.getMessage());
28.         }
29.     }

复制代码

（3）消耗者代码

1. @Test
2.     void contextLoads() {
3.         try {
4.             //1、建立连接
5.             ConnectionFactory factory = new ConnectionFactory();
6.             //2、设置连接参数
7.             factory.setHost("ip地址"); //主机名
8.             factory.setPort(5672); //连接端口
9.             factory.setVirtualHost("/"); //虚拟主机
10.             factory.setUsername("用户名"); //用户名
11.             factory.setPassword("密码"); //密码
12.             //3、建立连接
13.             Connection connection = factory.newConnection();
14.             //4、创建通道
15.             Channel channel = connection.createChannel();
16.             //5、创建队列并声明队列
17.             String queueName = "simple.queue"; //队列名称
18.             channel.queueDeclare(queueName,false, false, false, null);
19.             //6、订阅消息
20.             channel.basicConsume(queueName, true, new DefaultConsumer(channel){
21.                 @Override
22.                 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
23.                     //将字节流内容转为消息本体
24.                     String message = new String(body);
25.                     System.out.println("消费者接收到消息 --------- > " + message);
26.                 }
27.             });
28.             System.out.println("等待消息中。。。。");
29.             //关闭管道和连接
30.             channel.close();
31.             connection.close();
32.         }catch (Exception e) {
33.             System.out.println("生产者发送消息失败 ----- > " + e.getMessage());
34.         }
35.     }

复制代码

根本消息队列的消息发送流程：

• 建立connection
  
• 创建channel
  
• 利用channel声明队列
  
• 利用channel向队列发送消息
  

根本消息队列的消息接收流程

• 建立connection
  
• 创建channel
  
• 利用channel声明队列
  
• 界说consumer的消耗行为handleDelivery()
  
• 利用channel将消耗者与队列绑定
  

2、工作消息队列（WorkQueue）

3、发布订阅（Publish、Subscribe）模式

根据交换机的类型，可分为不同的三种
（1）广播（Fanout Exchange）

（2）路由（Direct Exchange）

（3）主题（Topic Exchange）

五、SpringAMQP

1、先容

AMQP：Advanced Message Queueing Protocol，是用于在应用步伐或之间传递业务消息的开放标准，该协议与语言平台无关，更符合微服务中独立性的要求。
SpringAMQP：Spring AMQP是基于AMQP协议界说的一套API规范，提供了模板来发送和接收消息，包括两部分，其中spring-amqp是基础抽象，spring-rabbit是底层默认实现。
2、示例

简单消息队列

发送消息：
（1）在父工程中引入依赖，确保每个子工程都能利用

1. <!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp -->
2. <dependency>
3.     <groupId>org.springframework.boot</groupId>
4.     <artifactId>spring-boot-starter-amqp</artifactId>
5. </dependency>

复制代码

（2）编写设置类

1. spring:
2.   rabbitmq:
3.     host: # 消息队列ip
4.     port: # 端口
5.     virtual-host: # 虚拟地址
6.     username: # 用户名
7.     password: # 密码

复制代码

（3）测试类

1. @SpringBootTest
2. class OrderServiceApplicationTests {
3.     @Autowired
4.     private RabbitTemplate rabbitTemplate;
5.     /**
6.      * 发送消息
7.      */
8.     @Test
9.     void sendMessageMQ() {
10.         //队列名称
11.         String queueName = "simple.queue";
12.         //消息
13.         String message = "你好，rabbitmq";
14.         //发送消息
15.         rabbitTemplate.convertAndSend(queueName, message);
16.     }
18. }

复制代码

接收消息：
（1）引入依赖

1. <!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp -->
2. <dependency>
3.     <groupId>org.springframework.boot</groupId>
4.     <artifactId>spring-boot-starter-amqp</artifactId>
5. </dependency>

复制代码

（2）编写设置类

1. spring:
2.   rabbitmq:
3.     host: # 消息队列ip
4.     port: # 端口
5.     virtual-host: # 虚拟地址
6.     username: # 用户名
7.     password: # 密码

复制代码

（3）接收并处置惩罚消息

1. @Component //监听消息的监听器
2. public class SpringRabbitListener {
3.         //需要监听哪个消息队列中的消息
4.     @RabbitListener(queues = "simple.queue")
5.     public void listenerSimpleQueue(String message) {
6.         //处理消息
7.         System.out.println("监听到 simple.queue 队列中的消息 --------- > " + message);
8.     }
9. }

复制代码

留意：消息一旦消耗就会从队列中移除，RabbitMQ没有消息回溯功能
Work Queue 工作队列

提高消息处置惩罚的速率，避免消息堆积
示例：
生产者，生产100条消息

1. @Test
2.     void sendMessageWorkQueue() throws InterruptedException {
3.         String queueName = "simple.queue";
4.         String message = "你好，rabbitmq__";
5.         for (int i = 1; i <= 100; i++) {
6.             rabbitTemplate.convertAndSend(queueName, message+i);
7.             Thread.sleep(20);
8.         }
9.     }

复制代码

两个消耗者，共同消耗，设置不同的处置惩罚速率

1. @Component
2. public class SpringRabbitListener {
4.     @RabbitListener(queues = "simple.queue")
5.     public void listenerWork1Queue(String message) throws InterruptedException {
6.         System.out.println("消费者1消费了消息 --------- > " + message + " " + LocalDateTime.now());
7.         Thread.sleep(20);
8.     }
10.     @RabbitListener(queues = "simple.queue")
11.     public void listenerWork2Queue(String message) throws InterruptedException {
12.         System.err.println("消费者2......消费了消息 --------- > " + message + " " + LocalDateTime.now());
13.         Thread.sleep(200);
14.     }
15. }

复制代码

结果：纵然设置了不同的处置惩罚速率，但是两个消耗者仍然处置惩罚了雷同的消息数量，并没有由于处置惩罚得快的消耗者会处置惩罚更多消息
原因：由于rabbitmq中的一个消息预取机制造成的
消息预取：当有大量消息到达消息队列时，队列会举行消息投递，消耗者的通道（channel）会预先把消息取出，就会均匀分配队列中的消息，并不会由于消耗者的处置惩罚能力举行分配
控制方法：
修改设置文件，设置值，可以控制预取消息的上限

1. spring:
2.   rabbitmq:
3.     host: # 消息队列ip
4.     port: # 端口
5.     virtual-host: # 虚拟地址
6.     username: # 用户名
7.     password: # 密码
8.     listener:      simple:        prefetch: 1 # 每次只能取1条消息，处置惩罚完后再取

复制代码

发布（Publish）、订阅（Subscribe）

发布订阅模式答应将同一消息发送给多个消耗者。实现方式是加入了exchange（交换机）。

留意：exchange（交换机）负责消息路由，而不是存储，路由失败则消息丢失
发布订阅-Fanout Exchange（广播）

Fanout Exchange会将接收到的消息路由到每一个跟其绑定的queue（队列）

示例：
（1）在消耗者中声明，将交换机与队罗列行绑定
声明两个队列和一个交换机，举行绑定

1. @Configuration
2. public class FanoutConfig {
3.     /**
4.      * 声明交换机
5.      * @return
6.      */
7.     @Bean
8.     public FanoutExchange fanoutExchange() {
9.         //返回一个声明的交换机
10.         return new FanoutExchange("fanout.exchange");   
11.     }
13.     /**
14.      * 声明第一个队列
15.      * @return
16.      */
17.     @Bean
18.     public Queue fanoutQueue1() {
19.         return new Queue("fanout.queue1");
20.     }
22.     /**
23.      * 声明第二个队列
24.      * @return
25.      */
26.     @Bean
27.     public Queue fanoutQueue2() {
28.         return new Queue("fanout.queue2");
29.     }
31.     /**
32.      * 绑定队列到交换机
33.      * @param fanoutQueue1
34.      * @param fanoutExchange
35.      * @return
36.      */
37.     @Bean
38.     public Binding bindingFanoutQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
39.         //将队列1绑定到交换机
40.         return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
41.     }
42.     /**
43.      * 绑定队列到交换机
44.      * @param fanoutQueue2
45.      * @param fanoutExchange
46.      * @return
47.      */
48.     @Bean
49.     public Binding bindingFanoutQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange) {
50.         //将队列2绑定到交换机
51.         return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
52.     }
53. }

复制代码

（2）处置惩罚消息

1. @Component
2. public class SpringRabbitListener {
3.     @RabbitListener(queues = "fanout.queue1")
4.     public void listenerFanoutQueue1(String message) {
5.         System.out.println("监听到 fanout.queue1 队列中的消息 --------- > " + message);
6.     }
8.     @RabbitListener(queues = "fanout.queue2")
9.     public void listenerFanoutQueue2(String message) {
10.         System.out.println("监听到 fanout.queue2 队列中的消息 --------- > " + message);
11.     }
12. }

复制代码

（3）发送消息

1. /**
2. * 发送消息到交换机
3. */
4. @Test
5. void testSendFanoutExchange() {
6.     String exChangeName = "fanout.exchange";
7.     String message = "交换机转发的消息";
8.     rabbitTemplate.convertAndSend(exChangeName, "", message);
9. }

复制代码

这样就实现了每个队列都能消耗消息，而不是消息一旦被其中一个消耗者消耗后，就会移除的题目
发布订阅-DirectExchange（路由模式）

Direct Exchange会将接收到的消息根据规则路由到指定的Queue，因此称为路由模式（routes）。

• 每一个Queue都会与Exchange设置一个BindingKey
  
• 发布者发送消息时，指定消息的RoutingKey
  
• Exchange将消息路由到BindingKey与消息RoutingKey一致的队列
  
  
  
  示例：
  

（1）声明队列，交换机，通过注解的方式

1. @Component
2. public class SpringDirectListener {
3.     /**
4.      * 声明第一个消费者
5.      * @param message
6.      */
7.     @RabbitListener(bindings = @QueueBinding(
8.             value = @Queue(name = "direct.queue1"),
9.             exchange = @Exchange(name = "direct.exchange", type = ExchangeTypes.DIRECT),
10.             key = {"red", "blue"}
11.     ))
12.     public void listenerDirect1Queue(String message) {
13.         System.out.println("direct消费者1监听到的消息 ---------- > " + message);
14.     }
16.     /**
17.      * 声明第二个消费者
18.      * @param message
19.      */
20.     @RabbitListener(bindings = @QueueBinding(
21.             value = @Queue(name = "direct.queue2"),
22.             exchange = @Exchange(name = "direct.exchange", type = ExchangeTypes.DIRECT),
23.             key = {"red", "yellow"}
24.     ))
25.     public void listenerDirect2Queue(String message) {
26.         System.out.println("direct消费者2监听到的消息 ---------- > " + message);
27.     }
28. }

复制代码

（2）发消息

1. /**
2. * 发送消息到交换机
3. */
4. @Test
5. void testSendDirectExchange() {
6.     String exChangeName = "direct.exchange";
7.     String message = "direct交换机转发的消息";
8.     String routingKey = "red"; //发送给指定routingKey的队列
9.     rabbitTemplate.convertAndSend(exChangeName, routingKey, message);
10. }

复制代码

发布订阅-TopicExchange（主题模式）

TopicExchange与DirectExchange类似，区别在于routingKey必须是多个单词的列表，并且用 . 分割

示例：
（1）声明交换机，队列

1. @Component
2. public class SpringTopicListener {
4.     @RabbitListener(bindings = @QueueBinding(
5.             value = @Queue("topic.queue1"),
6.             exchange = @Exchange(value = "topic.exchange", type = ExchangeTypes.TOPIC),
7.             key = {"china.#"}
8.     ))
9.     public void listenerTopic1Queue(String message) {
10.         System.out.println("Topic消费者1监听到的消息 ---------- > " + message);
11.     }
13.     @RabbitListener(bindings = @QueueBinding(
14.             value = @Queue("topic.queue2"),
15.             exchange = @Exchange(value = "topic.exchange", type = ExchangeTypes.TOPIC),
16.             key = {"#.news"}
17.     ))
18.     public void listenerTopic2Queue(String message) {
19.         System.out.println("Topic消费者2监听到的消息 ---------- > " + message);
20.     }
21. }

复制代码

（2）测试消息发送

1. /**
2. * 发送消息到交换机
3. */
4. @Test
5. void testSendTopicExchange() {
6.     String exChangeName = "topic.exchange";
7.     String message = "topic交换机转发的消息";
8.     String routingKey = "11.news";
9.     rabbitTemplate.convertAndSend(exChangeName, routingKey, message);
10. }

复制代码

六、消息转换器

测试：
（1）新增一个队列（queue）

1. /**
2. * 声明一个队列
3. * @return
4. */
5. @Bean
6. public Queue objectQueue() {
7.     return new Queue("object.queue");
8. }

复制代码

（2）测试发送对象类型数据

1. /**
2. * 发送消息
3. */
4. @Test
5. void sendObjectMessageMQ() {
6.     String queueName = "object.queue";
7.     Map<String, Object> map = new HashMap<>();
8.     map.put("username", "张三");
9.     map.put("age", 12);
10.     rabbitTemplate.convertAndSend(queueName, map);
11. }

复制代码

结果：

原因：
rabbitmq只支持字节，而spring答应发object对象，说明spring会将对象做序列化，用的序列化方式是jdk的序列化方式x-java-serialized-object，这种序列化缺点是性能差，安全性有题目，可能存在注入风险，数据长度太长了
Spring的对象处置惩罚事由org.springframework.amqp.support.converter.MessageConverter来处置惩罚的，而默认实现的SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。如果要修改必要界说一个MessageConverter类型的Bean即可。
步调：
（1）导入依赖

1. <!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.dataformat/jackson-dataformat-xml -->
2. <dependency>
3.     <groupId>com.fasterxml.jackson.dataformat</groupId>
4.     <artifactId>jackson-dataformat-xml</artifactId>
5.     <version>2.17.1</version>
6. </dependency>

复制代码

（2）声明一个bean，将消息转换的bean注入spring容器中

1. /**
2. * 声明消息转换类型
3. * @return
4. */
5. @Bean
6. public MessageConverter messageConverter() {
7.     return new Jackson2JsonMessageConverter();
8. }

复制代码

（3）将消息清除，重新测试

消息接收：
（1）引入依赖

1. <!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.dataformat/jackson-dataformat-xml -->
2. <dependency>
3.     <groupId>com.fasterxml.jackson.dataformat</groupId>
4.     <artifactId>jackson-dataformat-xml</artifactId>
5.     <version>2.17.1</version>
6. </dependency>

复制代码

（2）注入bean

1. /**
2. * 声明消息转换类型
3. * @return
4. */
5. @Bean
6. public MessageConverter messageConverter() {
7.     return new Jackson2JsonMessageConverter();
8. }

复制代码

（3）消息接收处置惩罚

1. @RabbitListener(queues = "object.queue")
2. public void listenerObjectQueue(Map<String, Object> map) {
3.     System.out.println("监听到 object.queue 队列中的消息 --------- > " + map);
4. }

复制代码

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