消息队列Message Queue 0基础学习

心已赠人 2024-03-16 23:01 72阅读 0赞

### 一、定义 ###

*  消息队列：一般我们会简称它为MQ(Message Queue)。
 *  Message Query（MQ），消息队列中间件，很多初学者认为，MQ通过消息的发送和接受来实现程序的异步和解耦，mq主要用于异步操作，这个不是mq的真正目的，只不过是mq的应用，mq真正的**目的是为了通讯**。
 *  他屏蔽了复杂的通讯协议，像常用的dubbo，http协议都是同步的。 这两种协议很难实现双端通讯，A调用B,B也可以主动调用A，而且不支持长连接。mq做的就是在这些协议上构建一个简单协议——**生产者、消费者模型**，mq带给我们的不是底层的通讯协议，而是更高层次的通讯模型。他定义了两个对象：发送数据的叫做生产者，接受消息的叫做消费者，我们**可以无视底层的通讯协议，我们可以自己定义生产者消费者**。

先复习一下**队列**。

*  队列是一种先进先出的数据结构（线性结构）。和他相似的还有栈（先进后出）。
 *  使用队列的还有一个场景就是线程池队列：有界队列和无界队列。用来存储任务。

### 二、使用场景 ###

消息队列的三个最主要的**作用**是：**异步、削峰、解耦**，一定要烂熟于心。

#### 传统设计 ####

先看看传统的设计（如果说业务体量小，并发不大，可以使用这种设计）

![b73156ff8d904fc8979b7f96a0dbd0ba.png][]

观察一下上面的设计，属于典型的**串行化调用，**这种设计模式有一个很大的优势，就是代码简单，出现问题很容易定位到问题。但是也有很多劣势，下面咱们从三高（高并发，高性能，高可用）三个方面去评审一下这个设计。

**高可用**：这些服务假如有一个服务挂掉（宕机或者网络波动），就意味着我这个请求失败了，这样用户体验会极差，用户会频繁看到支付失败。

**高并发**：因为这些操作都是由一个线程（主线程）去执行这些操作，所以当我们的QPS如果很高的话，很容易造成超时。

> QPS：系统每秒钟收到的请求。

**高性能**：因为上面这种设计模式是串行的，假设我的每次网络传输耗时200ms，业务处理需要20ms，完成上面那些操作需要耗时2s，这样用户体验也会很差（想象一下每次下单都需要等2s）,如果用户下单后的操作越来越多，耗时只会越来越高。

所以在一个大型的互联网项目中，以上设计是完全不可取的（非核心模块除外）。

#### 并行处理调优 ####

既然上面说的是串行模块，那么我们用自己的线程池把他改为**并行**的设计，再看评审一下。

所谓的并行设计就是原来由一个线程去串行做的逻辑，改为**多个线程并行**去做。

![2ed02bc1ea9241198d3307bb4bd2f125.png][]

**高可用**：这些服务假如有一个服务挂掉（宕机或者网络波动），理论上讲，如果补偿服务做的出色的话，还是满足高可用的。（可以用try,catch）

**高并发**：相比上面的设计，系统的吞吐量可以达到了很大程度上的提升。

**高性能：**相比上面的设计，因为很多业务是并行执行的，所以相当于只有200\*2+20，就可以返回。

> 上面这个设计看起来还是不错的设计，所以在很多这种串行调用，多次io的时候，我们就可以采用这种方案，上面这种设计也是多线程的一种实战应用。

下面来分析一下**弊端**：

1.系统的可扩展性太差了。上面只是列举了4步，但是实际上会有几十步，这几十步放到代码里就会像屎堆一样，可维护性极差。每次加一个步骤，都要多调一个接口，然后重新发布一下服务。

2，系统的耦合性太高了。想象一下，几十个http调用放到一起并发执行，很有可能会影响其他的点，尤其是淘宝京东这种秒杀敏感的业务，和钱挂钩的业务，很容易出现p0级别的bug。

3，使用的业务本身的线程池，在并发很多的情况下，容易造成cpu的竞争。

于是就有了进一步演化。

#### 多线程处理三种方案： ####

1，主线程交给线程池里的线程去执行时，主线程等待，等我所有任务，执行完之后，主线程再去接着执行（查询做聚合的时候）

2，主线程交给线程池去处理，就不管了（我要删除或修改一个东西，需要耗时2s）

3，主线程交给线程池去处理，主线程继续去执行其他的任务，等线程池里的任务执行完之后，去通知主线程，主线程在接着去执行后续的逻辑。

#### 消息队列 ####

![41d8171ef81e4320a7155705e30d6cf7.png][]

咱们从继续从三高的层面去审视一下这个设计：

**高可用：**当我系统里的一个模块宕机了，不会影响到我其他服务。（可以通过数据补偿或者分布式事务来保证数据最终一致性）

**高性能：**用户下单，将下单所需要的数据都放到消息队列里，就直接返回了，所有耗时相当于就是网络传输所耗时。

**高并发：**由于消息队列不处理任何业务上的逻辑，所有他支持的并发是百万级别的。假如有100万个用户下单，100万的数据放到消息队列里，连接消息队列的服务慢慢消费即可，也不至于造成瞬间有百万请求进来，将我的服务压垮。

消息队列的一些术语：

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>producer</td> 
   <td>消息生产（发布）者</td> 
  </tr> 
  <tr> 
   <td>consumer</td> 
   <td>消息消费（订阅）者</td> 
  </tr> 
 </tbody> 
</table>

*  ##### 三大优点： #####

1.  **解耦**：就像高可用里面说的一样，发淘金币服务挂了，关下单什么关系，发淘金币服务挂了，我还是可以正常下单，只不过后期可以数据补偿或者分布式事务去解决这个问题。
2.  **削峰**：比如说我平时服务就只能支撑几万的qps，像淘宝京东那种秒杀，那时候服务突然打进来（如果采用第二种方案）那服务就会直接被压死了。但是如果采用消息队列，这秒杀进来的所有的请求都不会直接打到具体服务上，都会先打到消息队列里，然后我后面的服务再慢慢消费。  
    可以看看淘宝京东双11秒杀的时候，是不是有的时候慢是慢了点，但是服务起码没挂。等我秒杀结束之后，服务还能正常运转。  
    消息队列就像是一个三峡大坝，用来拦截上游给的压力。
3.  **异步：**连接消息队列的服务可以异步去执行。而且每次多增加一个步骤，我下单的代码是不需要动的，只需要再增加一个消费者即可。

*  ##### 缺点 #####

1.  增加了系统复杂性。  
    所以说如果说你的业务量不大，并发也不高，就没必要使用消息队列。
2.  事务问题。  
    事务问题其实是分布式系统肯定会存在的一个问题，只不过消息队列更严重一些。一般解决方案有两种，第一种就是采用分布式事务，这个下单的里涉及的所有服务放到一个事务里面，要么都成功，要么都失败。第二种就是，消费者做好合理的数据补偿措施，比如说，消息重试，人工刷数据等等。
3.  可用性  
    刚才讲了解耦，其实是系统的各个模块之间的解耦，但是这些模块都和消息队列关联，万一消息队列挂了，就真的下不了单了。为了保证可用性，我们可以采用消息队列集群，前端流量限流等，后面会介绍。

#### 几种常见的消息队列 ####

![5c6053aee5ae4b51b4c5b2990c76529c.png][]

其中**kafka**和**rocketmq**现在在互联网大厂使用最广泛，因为吞吐量大。

##### MQ的两种流派 #####

> 什么是broker？可以理解为是一个中转站。生产者将消息发送给他就结束自己的任务了，broker将消息主动推送给消费者（具体的将消息推送到哪个队列，或者说消费者主动请求）

*  **有broker的**

重topic：必须要有topic

1.  kafka：全球消息处理性能最快的一款mq
2.  rocketmq：阿里内部的一个大神根据kafka的执行原理手写的，性能与kafka差不多，但是功能上比kafka要多，比如说顺序消费。

轻topic：可以没有topic，topic只是一种中转模式

1.  rabbitmq

*  **无broker的**

1.  zeromq：没有使用broker，是直接使用socket进行通信。

### MQ常见的问题 ###

#### 1、mq如何避免消息堆积问题。 ####

*  消息堆积：生产者的生产速率远远大于消费者的消费速率，使消息大批量的堆积在消息队列。
 *  解决方案：

1.  提升消费者的消费速率（增加消费者集群）
2.  消费者分批多线程去处理
3.  限流，保证进入到消息队列的都是有用的消息

#### 2、如何避免重复消费问题 ####

*  **产生原因：**

1.  生产者产生了两条一模一样的消息。
2.  消费者一条消息消费了多遍

*  **消息重试：**消息重试一般发生于一个消费者发生了异常（网络波动或者系统假死），这个时候这个消费者就会通知生产者重新发送。就会带来重复消费的问题。  
    可以采用常用的**幂等解决方案（分布式锁）**，全局id+业务场景保证唯一性。所有的重复提交问题，都可以用幂等性来解决。
 *  为了保险起见，也可以在数据库上做好**唯一索引**。

#### 3，如何保证消息不丢失 ####

1，消息确认机制，生产者必须确认消息成功刷盘到硬盘中，才确认消息发送成功。

acks 参数指定了必须要有多少个分区副本收到消息，生产者才会认为消息写入是成功的。这个参数对消息 丢失的可能性有重要影响。该参数有如下选项。

*  如果 acks=0 ，生产者在成功写入消息之前不会等待任何来自服务器的响应。也就是说，如果当中出现了问题，导致服务器没有收到消息，那么生产者就无从得知，消息也就丢失了。不过，因为生产者不需要等待服务器的响应，所以它可以以网络能够支持的最大速度发送消息，从而达到很高的吞吐量。
 *  如果 acks=1 ，只要集群的Leader节点收到消息，生产者就会收到一个来自服务器的成功响应。如果消息无法到达Leader节点（比如Leader节点崩溃，新的Leader还没有被选举出来），生产者会收到一个错误响应，为了避免数据丢失，生产者会重发消息。不过，如果一个没有收到消息的节点成为新Leader，消息还是会丢失。这个时候的吞吐量取决于使用的是同步发送还是异步发送。如果让发送客户端等待服务器的响应（通过调用 Future 对象的 get() 方法），显然会增加延迟（在网络上传输一个来回的延迟）。如果客户端使用回调，延迟问题就可以得到缓解，不过吞吐量还是会受发送中消息数量的限制（比如，生产者在收到服务器响应之前可以发送多少个消息）。
 *  如果 acks=-1(或all)，只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。这种模式是最安全的，它可以保证不止一个服务器收到消息，就算有服务器发生崩溃，整个集群仍然可以运行。不过，它的延迟比 acks=1 时更高，因为要等待不只一个服务器节点接收消息。

结合具体的业务场景来进行选择。

2，消息持久化机制，作为mq中间件，会把消息持久化到硬盘，因为内存里的数据断电就丢失了

3，消费者必须确认消息消费成功，否则进行重试，重试达到一定次数后，通知开发人员做好补偿措施。

> 关闭自动提交，当消费者消费的完成后，再手动提交，防止mq的自动提交，当消费者接收到消息后，mq以为消费者已经消费了，但这个时候消费者如果挂了，这条消息就没有被消费。

#### 4，如何保证消费的顺序一致性 ####

大多数的项目不需要保证顺序一致性，某些特殊场景必须保证顺序一致性，比如说，mq用于保证redis和mysql的数据一致性。

绑定同一个消费队列，消费的时候进行要注意如果使用了多线程处理，避免重新创建list，要在原来的list进行修改。

#### 5，mq怎么用于保证redis和数据库的数据一致性 ####

1，当执行update后，发送mq去通知消费者更新redis数据

*  优点:解耦，提高接口响应速度，有相应的补偿策略
 *  缺点：延迟比较高

2，监听binlog日志，结合mq，去更新redis（canal实现）

*  优点：更加解耦
 *  缺点：延迟更高

3，双删策略

#### 6，消费者怎么知道mq里有消息了 ####

两种方案：

1，mq主动通知（push）

当mq中有消息，就会通知消费者来进行消费。这种模型有一个致命伤，就是**慢消费**。

2，消费者轮询（pull）

消费者去轮询看看有没有自己要消费的消息。这种模型也有弊端就是消息延迟与忙等。

如果消费者的速度比发送者的速度慢很多，势必造成消息在mq的堆积。假设这些消息都是有用的无法丢弃的，消息就要一直在mq端保存。当然这还不是最致命的，最致命的是mq给消费者推送一堆无法处理的消息，消费者不是拒绝就是报错，然后来回踢皮球。

反观pull模式，消费者可以按需消费，不用担心自己处理不了的消息来骚扰自己，而mq堆积消息也会相对简单，无需记录每一个要发送消息的状态，只需要维护所有消息的队列和偏移量就可以了。所以消息量有限且到来的速度不均匀的情况，pull模式比较合适。

由于主动权在消费方，消费方无法准确地决定何时去拉取最新的消息。如果一次pull取到消息了还可以继续去pull，如果没有pull取到则需要等待一段时间重新pull。

但等待多久就很难判定了。当然也不是说延迟就没有解决方案了，业界较成熟的做法是从短时间开始（不会对mq有太大负担），然后指数级增长等待。比如开始等5ms，然后10ms，然后20ms，然后40ms……直到有消息到来，然后再回到5ms。

即使这样，依然存在延迟问题：假设40ms到80ms之间的50ms消息到来，消息就延迟了30ms，而且对于半个小时来一次的消息，这些开销就是白白浪费的。

在阿里的RocketMq里，有一种优化的做法-长轮询，来平衡推拉模型各自的缺点。基本思路是:消费者如果尝试拉取失败，不是直接返回,而是把连接挂在那里等待,服务端如果有新的消息到来，把连接复用起来，这也是不错的思路。但海量的长连接mq对系统的开销还是不容小觑的，还是要合理的评估时间间隔。

#### 7，如果mq宕机后，生产者怎么处理。 ####

生产者在向mq投递消息的时候，可以将要投递的消息记录下来（可以在数据库中插入一条数据，也可以输出相应的日志记录）后期可以编写定时任务，定期向mq发送之前发送不成功的消息。

#### 8，mq的消费策略 ####

*  集群消费：同一个消费者集群，只能消费一条消息，但是一条消息可以被多个消费者集群消费。
 *  广播消费：通知集群中的所有节点都进行消费（涉及到数据分片处理的场景），对数据不敏感 的场景可以采用普通hash，对数据敏感的场景可以采用**hash环**。

[b73156ff8d904fc8979b7f96a0dbd0ba.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/e52113231b5946338ff0217899600867.png
[2ed02bc1ea9241198d3307bb4bd2f125.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/e242de0acf4d4e6bba4d209abae23a62.png
[41d8171ef81e4320a7155705e30d6cf7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/4310d2c2edf444dd8f86221ce9eea8f4.png
[5c6053aee5ae4b51b4c5b2990c76529c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/e24478b768744a5e9af4263255485c3c.png