微服务的异步通信技术RabbitMQ

小灰灰 2023-09-23 23:11 15阅读 0赞

#### 文章目录 ####

*  前言
 *   *  1.WorkQueue（工作队列）
     *   *  消息预取机制
     *  2.Publish&Subscribe（发布-订阅）
     *   *  1.Fanout（广播）
         *  2.DirectExchange（路由）
         *  3.TopicExchange（话题）
     *  MQ的优点

## 前言 ##

MQ的出现进一步降低了微服务模块之间的耦合度，相比于同步通信而言减少了关联服务的等待时间，使消息的传递更加多变，灵活  
**不管什么东西，只要被Spring整合就会变得十分简单，RabbitMQ也不例外**  
使用SpringAMQP来实现消息收发，不需要重复地配置连接参数，解决了一部分“硬编码”的问题。可以说和MyBatis整合JDBC非常相似。  
在以前，使用原生的RabbitMQ收发消息是这样的：  
![5311a93d577f4d8899d261e8de5afa6e.png][]  
使用SpringAMQP后收发消息是这样的：  
这就是一个基本队列（Basic-Queue）  
![在这里插入图片描述][d107cc6e9f6944b4b36f8f2cb8baf531.png]  
可以看到，只要引入依赖`spring-boot-starter-amqp`，写好yml配置文件，建立连接、创建通道的工作Spring都为我们做好了，而我们要做的仅仅就是利用工具类发送、监听消息，可以说相当的方便！  
针对不同的场景，我们要使用不同的队列模型：

### 1.WorkQueue（工作队列） ###

![在这里插入图片描述][27f5b9e91d0a445ca6e2f38b6c62509f.png]  
对于单一消费者情况（简单队列），当生产者每秒发送50条消息，消费者每秒处理40条消息，这样每秒钟就会多出10条消息无法处理，由此就会产生生产过剩而导致消息堆积在队列中，一旦达到队列内存的上限，新来的消息就无法被处理而被丢弃。  
为了提高消息处理的速度，避免队列中消息的堆积可以将队列绑定多个消费者，即WorkQueue  
![在这里插入图片描述][c0c85d0019f74205a6ae84adb27f4658.png]  
为了方便观察控制台，一般这样设计：  
生产者：

@Test
    public void testSendMessage2WorkQueue() throws InterruptedException {
        
        String queueName = "work.queue";
        String message = "hello, MQQ";
        for (int i = 1; i <= 50; i++) {
        
            rabbitTemplate.convertAndSend(queueName, message + i);
            Thread.sleep(20);
        }
    }

消费者：  
为了尽可能的模拟真实场景（消费者处理消息的能力不同），所以设置两个消费者的sleep参数为不同的两个时间

@RabbitListener(queues = "work.queue")
    public void listenWorkQueue1(String msg) throws InterruptedException {
        
        System.out.println("消费者1接收到消息——【" + msg + "】" + "At "+LocalTime.now());
        Thread.sleep(20);
    }
    
    @RabbitListener(queues = "work.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        
        System.err.println("消费者2接收到消息——【" + msg + "】" + "At "+ LocalTime.now());
        Thread.sleep(200);
    }

#### 消息预取机制 ####

运行后观察控制台，发现所有的消息处理完时间竟然花费了差不多五秒钟，很显然这样的效率是非常低下的：  
![在这里插入图片描述][280ca24f483242a2b0cd0747a8ea8a68.png]  
**为何绑定了两个消费者消费消息的速度不快反慢了呢？**  
仔细观察控制台会发现50条消息是被平均分配的，两个消费者分别消费id为偶数、奇数的消息，这样一想好像是**处理能力差的消费者拖了后腿(200msX25=5000ms=5s)**，为什么会出现这样的情况？  
这是由于MQ存在消息预取机制，即消费者会在处理之前预先拿到消息的通道，然后逐个处理消息，这个过程是与处理消息相隔离的！  
如果还有人不明白就想想使用原生的RabbitMQ时我们是怎么处理消息的：  
![在这里插入图片描述][32bc35e8ab034f2eba4531a06fa4da4c.png]  
当执行完回调函数有可能消息都不会被处理，这时程序会继续向下执行，过段时间才会开始处理消息（其实我认为**这也是体现RabbitMQ异步的一个地方**）  
这样的机制存在是保证异步性的关键，通过人为的设置参数也可以将消息预取的方式做出调整，来保证处理的效率，就像这样：

listener:
      simple:
        prefetch: 1

通过prefetch参数来保证消费者每次获取消息的个数，以及处理完成后才能获取下一个批次的消息  
进行数据预取设置后消费者在一秒之内处理完了所有的消息：  
![在这里插入图片描述][f6666722ff564149bdfb55f41f1e869b.png]  
**由此可见对于WorkQueue中消费者的设置要进行“按劳分配”的策略才较为完美  
使用工作队列WorkQueue之后处理消息的效率得到了很大的提升，并且也不会出现消息堆积的情况**

### 2.Publish&Subscribe（发布-订阅） ###

对于简单队列和工作队列模型，生产者发布消息，消费者一旦消费完，消息就会被销毁。这样无法做到将一个消息同时发送给多个消费者。  
![在这里插入图片描述][4d8ab1bd9684438b84f8e80588f96c3a.png]  
**对于一个微服务项目，在支付订单的模型中当支付服务完成，会发消息同时去通知短信服务、订单服务…这就需要保证消息的高可用，不能一个服务消费完消息就被销毁而导致其他服务接收不到消息**  
如何做到将同一消息发送给多个消费者并让其各自接收到？采用发布&订阅的工作模型即可  
![在这里插入图片描述][fdc6fecab01d4d19a8740060ad439cdb.png]  
**通过交换机（exchange）将消息路由到不同的队列中，再由消费者来消费各自订阅队列中的消息**  
针对不同的交换机种类会有不同的发布策略：

#### 1.Fanout（广播） ####

SpringAMQPA提供了声明交换机、队列、绑定关系的API  
![在这里插入图片描述][312afbe3bdd04c7cb4692b012b465e00.png]  
所以使用Exchange接口下的实现类就可以实现将消息路由到每一个绑定的Queue中，使用代码就会变得非常简单，声明交换机并绑定队列即可：

@Bean
    public FanoutExchange fanoutExchange(){
        
        return new FanoutExchange("yu7.fanout");
    }
    
    // fanout.queue1
    @Bean
    public Queue fanoutQueue1(){
        
        return new Queue("fanout.queue1");
    }
    
    // 绑定队列1到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
        
        return BindingBuilder
                .bind(fanoutQueue1)
                .to(fanoutExchange);
    }
    
    // fanout.queue2
    @Bean
    public Queue fanoutQueue2(){
        
        return new Queue("fanout.queue2");
    }
    
    // 绑定队列2到交换机
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
        
        return BindingBuilder
                .bind(fanoutQueue2)
                .to(fanoutExchange);
    }

![在这里插入图片描述][41af8046619144c38fdcbf2e15fa403a.png]  
注意：交换机只能用作消息的转发路由，不能用作消息的存储，一旦路由失败消息就会丢失！

#### 2.DirectExchange（路由） ####

DirectExchange会将接收到的消息根据规则路由到指定的Queue中，生产者发布消息时指定消息的RoutingKey与消费者声明的bindingKey相匹配，从而达到“精确制导”  
![在这里插入图片描述][7fbf32f1ab0c477c8a06cf658d896315.png]  
为了简化开发不使用声明Bean的方式来完成配置，通过@RabbitListener注解即可一键完成，所以根本不需要使用配置类：

生产者：

@Test
    public void testSendDirectExchange() {
        
        // 交换机名称
        String exchangeName = "yu7.direct";
        // 消息
        String message = "hello, MQ!";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "A", message);
    }

消费者：

@RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name = "yu7.direct", type = ExchangeTypes.DIRECT),
            key = {
        "A", "B"}
    ))
    public void listenDirectQueue1(String msg){
        
        System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】");
    }
    
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(name = "yu7.direct", type = ExchangeTypes.DIRECT),
            key = {
        "A", "C"}
    ))
    public void listenDirectQueue2(String msg){
        
        System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】");
    }

![在这里插入图片描述][b921c4fe677f487087b987597256af00.png]  
**当队列的bindingKey都相同时就变成了广播模型**

#### 3.TopicExchange（话题） ####

Topic与Direct非常相似，他允许RoutingKey-BindingKey以通配符的形式进行匹配，这样就可以更加有针对性的路由、订阅更多消息，并且以前用多个BindingKey的情况现在只需要用一个就能解决：  
\#：代表0或者多个单词  
\*：代指一个单词  
![在这里插入图片描述][0e2d3e55e5c54e85a0b7e612ade3fe45.png]

### MQ的优点 ###

1、耦合度低:每次有新需求，只需要添加对应的订阅即可  
2、吞吐量提升:各自处理自己订阅的事件，不需要等待执行完毕后再释放资源  
3、故障隔离:因为没有强依赖，中间某一环节出了问题，不会影响整个流程  
4、流量削峰:MQ就像—根管道，大量请求来了，你们给我排好队，依次执行

[5311a93d577f4d8899d261e8de5afa6e.png]: https://img-blog.csdnimg.cn/5311a93d577f4d8899d261e8de5afa6e.png
[d107cc6e9f6944b4b36f8f2cb8baf531.png]: https://img-blog.csdnimg.cn/d107cc6e9f6944b4b36f8f2cb8baf531.png
[27f5b9e91d0a445ca6e2f38b6c62509f.png]: https://img-blog.csdnimg.cn/27f5b9e91d0a445ca6e2f38b6c62509f.png
[c0c85d0019f74205a6ae84adb27f4658.png]: https://img-blog.csdnimg.cn/c0c85d0019f74205a6ae84adb27f4658.png
[280ca24f483242a2b0cd0747a8ea8a68.png]: https://img-blog.csdnimg.cn/280ca24f483242a2b0cd0747a8ea8a68.png
[32bc35e8ab034f2eba4531a06fa4da4c.png]: https://img-blog.csdnimg.cn/32bc35e8ab034f2eba4531a06fa4da4c.png
[f6666722ff564149bdfb55f41f1e869b.png]: https://img-blog.csdnimg.cn/f6666722ff564149bdfb55f41f1e869b.png
[4d8ab1bd9684438b84f8e80588f96c3a.png]: https://img-blog.csdnimg.cn/4d8ab1bd9684438b84f8e80588f96c3a.png
[fdc6fecab01d4d19a8740060ad439cdb.png]: https://img-blog.csdnimg.cn/fdc6fecab01d4d19a8740060ad439cdb.png
[312afbe3bdd04c7cb4692b012b465e00.png]: https://img-blog.csdnimg.cn/312afbe3bdd04c7cb4692b012b465e00.png
[41af8046619144c38fdcbf2e15fa403a.png]: https://img-blog.csdnimg.cn/41af8046619144c38fdcbf2e15fa403a.png
[7fbf32f1ab0c477c8a06cf658d896315.png]: https://img-blog.csdnimg.cn/7fbf32f1ab0c477c8a06cf658d896315.png
[b921c4fe677f487087b987597256af00.png]: https://img-blog.csdnimg.cn/b921c4fe677f487087b987597256af00.png
[0e2d3e55e5c54e85a0b7e612ade3fe45.png]: https://img-blog.csdnimg.cn/0e2d3e55e5c54e85a0b7e612ade3fe45.png