RabbitMQ是什么？RabbitMQ简介

一：技术配景

           假如我们有一个支付服务，支付服务的业务逻辑是：起首支付扣减余额，更新支付单状态，更新订单状态，发短信，给这个用户增长积分。在这个场景下，如果我们利用同步调用通信，那么调用支付这个接口是不是得比及给这个用户增长积分的业务执行完后，这个调用链条依次返回到支付接口时，支付接口的业务才算完成，可见这个情况显着很不公道。起首支付接口等待时间太长了，我支付业务的核心逻辑做完后，还要分身着非核心的业务，那如果有一个非核心业务down掉了，或者因为网络题目或者硬件题目或者代码题目使这个接口不能响应或者响应时间很长，那么下面的业务也会做不了，这条线程就出了大题目。堆积在tomcat内，长期占用tomcat的资源，再有用户下单，照旧一样的占用tomcat的资源，现在一个服务失败了，业务中远程调用的接口及其的多，依次类推，就会造成服务级联失败，雪崩现象产生。假如我们来做限流，或者熔断，那肯定不行啊，这可是支付业务，非常告急，如果熔断，那么业务不可以，就可能造成数据不同等性等题目，这是非常严肃的。
  
          以是同步通信在有些场景下使不适用的，我们应该把业务的核心逻辑来做同步通信，（也就是，我们必须要拿到业务处置惩罚完成返回的响应，从而必须做下一步的操纵的这种场景下），在同步通信下，可做限流熔断等服务的掩护操纵；而非核心业务我们就用异步通信来做。非常常见的秒杀场景，它是一种流量激增的业务，激增的这段时间，qps及其高，如果采用同步调用，结果可想而知，轻则响应时间非常慢，用户体验差，重则服务器直接瓦解。异步调用就相当于把一个很长的链条拆开，留其告急的部分做同步调用，不太告急的做异步调用，就类似于把山峰削开，用来填山谷，让它变得平整。
  二：安装部署

官网：https://www.rabbitmq.com/tutorials/tutorial-one-java
基于docker安装（条件你已经在官网下载好了rabbitmq）
2-1：加载rabbitmq.tar

1. docker load -i mq.tar

复制代码

2-2： 运行rabbitmq

1. docker run \
2. -e RABBITMQ_DEFAULT_USER=rabbitmq \
3. -e RABBITMQ_DEFAULT_PASS=rabbitmq \
4. -v mq-plugins:/plugins \
5. --name mq \
6. --hostname mq \
7. -p 15672:15672 \
8. -p 5672:5672 \
9. --network zzb-network\
10. -d \
11. rabbitmq:3.8-management

复制代码

三：初识RabbltMQ
整体架构：

• virtual-host：假造主机，起到数据隔离的作用
  
• publisher：消息发送者
  
• consumer：消息的消费者
  
• queue：队列，存储消息
  
• exchange：互换机，负责路由消息
  

注意：互换机只能路由消息，无法存储消息；互换机只会路由消息给其绑定的队列，因此队列必须与互换机绑定
2-3 Java简单示例 (基于AMQP协议)

引入amqp-starter依靠

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

复制代码

设置MQ地址，在发送者服务和吸收者服务的application.yml中添加设置（最好设置在nacos中，方便管理）：

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

复制代码

发送者：

1. @Autowired
2. private RabbitTemplate rabbitTemplate;
3. @Test
4. public void send(){
5.     //队列名称
6.     String queue = "simple.queue";
7.     //消息
8.     String message = "hello RabbitMQ";
9.     //发送消息
10.     rabbitTemplate.convertAndSend(queue, message);
11. }

复制代码

接受者

1. // 把接受消息的类的对象放进ioc容器
2. @Component
3. @Slf4j
4. public class Listener {
5.    
6.     @RabbitListener(queues = "simple.queue")
7.     public void receive(String message) {
8.         log.info("接收到消息：{}", message);
9.     }
10. }

复制代码

三：SpringAMQP

3-1：Work queues，任务模型：

        Work queues，就是多个消费者监听同一个队列。
        在实际开发中，并不是在一个服务里编写多个消费者监听同一个队列，而是部署多个实例，形成集群，监听那个队列。
        1：Work queues的特点

        队列中的一条消息只会被一个消费者监听执行，如果有其他的消费者，是拿不到这个消息的；基于这个特点，如果有其他消费者，其他消费者就会监听执行这个队列后面的消息，以是对于一个队列中全部的消息，多消费者是可以更快的处置惩罚这些消息，以是加消费者是在高并发场景下一个不错的处置惩罚方案。
        我们要知道，部署多个实例，每个呆板的性能可能有好有坏，如果有台呆板性能比较差一些，那它处置惩罚消息的速率肯定比不外其他的消费者，那队列怎样分配消息给消费者呢？答案默认是轮询，每个消费者处置惩罚一个消息。显然这个方案和实际情况背道而驰，以是我们必要设置，让每一个消费者只能处置惩罚一条消息，等当前消息处置惩罚完毕后，才能获取下一条消息。这样完美解决了消息积蓄题目。在application.yaml中设置：

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

复制代码

        3-2：互换机

        互换机的作用重要是接受发送者发送的消息，并将消息路由到预期绑定的队列，常见的互换机类型有三种：
3-2-1：Fauout：广播rabbitTemplate.convertAndSend(x,x,x)三个参数

每个监听不同队列的消费者都可以接受到来自同一个互换机转发来的消息。
3-2-2：Direct：定向

Direct Exchange会将接受到的消息根据规则路由到指定的queue，因此称为定向路由

• 每一个queue都会erexchange设置一个BindingKey
  
• 发布者发送消息时，指定消息的RoutingKey
  
• Exchange将消息路由到BindingKey与消息RoutingKey同等的队列
  

3-2-3：Topic：话题

TopicExchange也是基于RoutingKey做消息路由，然是routingKey通常是多个单词的组合，并且以   .  分割，比如                 user.#                  item.#                     #.cn 
BindingKey可以利用通配符：#代表0个或者多个单词，*代表一个单词
        3-3 声明队列互换机

1. 基于Bean声明队列互换机

SpringAMQP提供了几个类，用来声明队列、互换机及其绑定关系:

• Queue:用于声明队别，可以用工厂类QueueBuilder构建
  
• Exchange:用于声明互换机，可以用工厂类ExchangeBuilder构建
  
• Binding:用于声明队列和互换机的绑定关系，可以用工厂类BindingBuilder构建
  

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("a.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. }

复制代码

direct示例：

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("a.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. }

复制代码

可以看到，声明direct互换机与队列绑定关系的代码显得非常臃肿，以是一般我们利用注解来声明

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

复制代码

3-4 消息转换器

发送消息时，默认的把消息转换是jdk来做的，把对象序列化为字节举行传输，这种方式不推荐
发起采用别的消息转换器，如json序列化
步骤一：在publisher和consumer两个服务中都引入依靠：

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依靠。
步骤二：设置消息转换器，在publisher和consumer两个服务的启动类中添加一个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. }

复制代码

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