RocketMQ的push消费方式实现的太聪明了

约定不等于承诺〃 2024-03-16 20:41 44阅读 0赞

### MQ消费方式 ###

消费方式就是指消费者如何从MQ中获取到消息，分为两种方式，push（推方式）和pull（拉方式）。

#### 1、push（推方式） ####

push，顾名思义，就是推的意思。就是当MQ收到生产者产生的消息的时候，会主动将消息推送到消费者进行消费，这种模式就叫push，也就是MQ将消息推给到消费者的意思。

![089ee5bc50205b1dc9784558cbd2c631.png][]

push模式

push这种模式的好处就是响应快，消息的实时性比较高，一旦消息MQ收到消息，那么就能立马将消息推送给消费者，消费者也就能立马收到消息进行消费。

但是这种push的模式，有个缺点就是一旦消息量比较大时，对消费者性能要求比较高，因为是消费者无法控制MQ消息的推送速度，一旦消息量大，那么消费者消费消息的压力就比较大。

#### 2、pull（拉方式） ####

push是MQ主动给消费者推消息，那么pull呢？刚好跟push相反，就是消费者主动去MQ中拉取消息。

![ca5d2f9eef141e086b564a767613107e.png][]

pull模式

那么pull的优缺点自然也就跟push刚好相反。因为是消费者主动去MQ中拉取消息，那么消费者可根据自身消费的情况，决定何时去拉取消息，主动权在自己手上，这样消费者的压力就会相对小点；但是缺点也很明显，那么就会实时性相对于push方式会低一些，因为你得决定拉的时间间隔。

其实想想，消费方式就跟拿快递一样，快递就是一个消息，我自己就是消费者，快递要么快递小哥主动送（push）到家，要么我自己去快递站拿（pull）。

### RocketMQ对于消费方式的实现 ###

上一节说了消费消息的两种方式push和pull，或者说算一种理念。尚大的周阳老师有一句经常说的话我比较赞同，那就是“天上飞的理念，必然有落地的实现”。所以push或者pull到底如何落地，得看具体的MQ的产品了。

而RocketMQ作为阿里开源的一款高性能、功能丰富的MQ，自然同时实现了push和pull的两种消费方式，用户可以选择在项目中使用push还是pull。

![c7bc92b369162562cf9e5c4c7c9d8f58.jpeg][]

push模式的实现pull模式的实现

但是一般情况下，项目中都是使用push的方式来消费，因为pull除了时实性差外，pull方式还得让开发人员主动去维护消息消费进度，增加额外的操作。

所以接下来就着重讲一下RocketMQ是如何实现push的逻辑。

### RocketMQ聪明地实现push的原因 ###

上文说到push模式的优点是时实性好，但是缺点就是消费者压力会比较大，所以，难道实现push模式，只能舍弃压力的控制么？

就在这时，RocketMQ大喊了一声

![46cfbb687a89adf7ec0af8ac69fb43f4.png][]

是的，RocketMQ对于push模式做到了实时和压力的平衡，这主要是因为RocketMQ的push模式其实算是一个“伪push”模式，真正底层的实现还是基于pull。

到这里可能有的小伙伴比较迷糊，怎么push变成“伪push”了，还是用pull实现的，到底是push还是pull？

![42111e2c22abfb39a3d49c0c0f85224f.jpeg][]

前面我说过，push和pull只是一种理论，具体的实现看MQ。

所以RocketMQ为了兼顾两者，就选择通过消费者主动拉消息来实现push的效果，这也是为什么我称为“伪push”的原因，RocketMQ都给封装好了，让你用起来感觉是MQ主动push消息给你的。

既然底层是pull，那么RokcetMQ在实现消费者的逻辑的时候，就可以很容易实现控制压力的效果，毕竟这是“拉”方式天然自带的buff；但是如何通过pull实现push的时实的优点呢？毕竟鱼和熊掌我RokcetMQ偏要兼得。

这时这就不得不提到一种叫“长轮询”的机制。

### 轮询与长轮询 ###

轮询与长轮询都属于pull的实现，都是由客户端主动给服务端发送请求，拉取数据。套到MQ中，就是都是消费者主动去MQ拉消息。

#### 轮询 ####

轮询是指不管服务端数据有无更新，客户端每隔定长时间请求拉取一次数据，可能有更新数据返回，也可能什么都没有。

![accf03c573f5e8d5cd48aa97d8ae808c.jpeg][]

再拿快递举例子，轮询就好比，小明买的iphone 13 pro max快递到了，显示正在派送中，但是小明等不及了，于是就去快递站拿，但是快递还没放到快递站，但是小明的心里急啊，他忍受不了相思之苦，于是小明每隔5分钟就往快递站跑一次，问一下快递到了没，到了就拿回来。这就是轮询的意思，也就是不论有没有数据，客户端都会每隔一定时间去请求一次服务端。

来分析一下拿快递的例子的问题：

*  每隔5分钟就往快递站跑，那不是累死个小明么。
 *  还有一个问题，假设刚跑到快递站，快递没到，就回去了，但是刚到家的时候，快递到了，于是又等了5分钟，再去快递站终于拿到快递了，但是其实快递都到了几分钟了，你还是没有第一时间拿到快递，这就造成了延迟。

从而对应到程序中，就是会产生如下问题

*  对于消息而言，会一直产生，这就要求消费者不停地间隔一定时间去拉取消息，即使没有消息也需要去请求，就会造成大量无用的请求，白白浪费大量耗费服务器内存和宽带资源。
 *  可能造成数据的延迟

#### 长轮询 ####

说长轮询概念之前，先来救救小明吧，毕竟小明可不想狗带。

既然原先小明每隔5分钟跑一次，那么是不是可以换种思路，当快递还没到的时候，让小明不要回来，直接在快递站待着，当快递到的时候，才让小明拿着快递回家。这下小明就喜死了，既可以有时间刷刷某音，逛逛某东，还可以在第一时间拿到13 pro max。

![cad0fd812bb0f92c479eb011021f88d0.jpeg][]

所以这种可以在快递站等待的机制，就叫长轮询。

长轮询也是客户端请求服务端，如果服务端有数据，那么就立马返回，客户端再次请求；当服务端不存在数据的时候，服务端并不会给客户端响应，而是将请求给hold住，当服务端有数据的时候才会给客户端响应，返回数据。

所以长轮询可以解决如下问题

*  解决轮询带来的频繁请求服务端但是没有的问题
 *  一旦新的数据到了，那么消费者能立马就可以获取到新的数据，所以从效果上，有点像是push的感觉。

但是长轮询也会带来服务端代码实现逻辑复杂的问题，当然相比于优点来说，都不太重要。

### push消费方式源码探究 ###

理论都讲完了，接下来就到了show me the code的时间了，来看看RocketMQ的是如何通过长轮询机制来实现压力和时实的平衡。

这里我画了一张push模式下消费者消费流程图。

![3b8fdb199f3a66ba077910d93e5266d7.jpeg][]

消费者拉取消息的逻辑

*  ①消费者有一个后台线程，会去处理拉取消息（PullRequest）
 *  ②先去判断有没有过多消息没有消费，如果有的话，那么就间隔一定时间再次从①开始执行拉取消息的逻辑
 *  ③消费者没有过多消息没有消费，那么就会直接向MQ发送拉取消息的请求，有消息就返回，没有消息就hold住请求，等有新的消息到的时候才返回
 *  ④消费者获取到消息之后，会去找用户自定义的消息处理逻辑的实现（MessageListener的实现）去消费消息，同时会再次拉取消息，继续从①开始执行逻辑

#### 1、消费者拉取消息控制压力源码 ####

当消费者准备去拉消息的时候，会先去判断当前消费者消费的压力再决定是否去拉取消息。

RocketMQ提供了两种判断消费压力逻辑，一种是基于还未消费的消息的数量的大小，还有一种是基于还未消费的消息所占内存的大小。

![d0971a0d08b97a0f893b175cc825b23e.png][]

控制压力源码

*  判断还未消费消息的数量，数量太多就等会再执行重新执行拉取消息的逻辑
 *  判断还未消费消息的大小，如果还未消息的消息占用的内存过大，就等会再执行重新执行拉取消息的逻辑

总的一句话就是，当消费者消费的压力过大时，就不会去拉取消息，而是等待一定的时间再去执行拉取消息的逻辑，如果压力还是很大，就还继续等，如此循环，直到消费者的消费压力小于阈值的时候，才会真正的发送请求到MQ中拉取消息。

#### 2、MQ将请求hold住源码 ####

当服务端未找到消息时，就将请求进行挂起，存起来

![c5a0b63b5c5b10df961f57591d77240d.png][]

请求hold住源码

拉取不到消息时，会调用PullRequestHoldService的suspendPullRequest方法讲请求存储起来。PullRequestHoldService是用来存储拉取请求的类。

![67ec50fedd944272403064f137488a4d.jpeg][]

PullRequestHoldService

suspendPullRequest方法会将请求分类，放到ManyPullRequest里，然后用一个ConcurrentHashMap进行存储

#### 3、MQ收到消息响应给消费者的源码 ####

![90043544b314add9685712f00b44d58f.png][]

NotifyMessageArrivingListener

当生产者发送的消息达到MQ的时候，MQ会回调NotifyMessageArrivingListener的arriving方法，之后就会调用PullRequestHoldService的notifyMessageArriving方法，MQ会重新处理拉取消息的逻辑，此时就能找到最新来的那条消息，从而将最新的消息通过网络返回给消费者。

![cef97ff66956c63d78c8599bd84efc04.jpeg][]

notifyMessageArriving和返回消息逻辑

### 最后 ###

所以从以上的分析可以看出，RocketMQ对于push的消费方式的实现是基于长轮询机制来实现的，同时平衡了时实和压力，这其实就很nice了。

最后我想说一句，其实不论是pull还是push，又或是轮询和长轮询，其实都是一种理论或者说是一种思想，不单单是MQ的东西，就比如在Nacos中，也使用了push和长轮询机制。但是这些理论在不同产品的具体实现，实现方式可能不太一样，但都是大同小异，所以当你懂了这些思想，再看其它框架的源码，其实就很容易了。

[089ee5bc50205b1dc9784558cbd2c631.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/dfe368f8700548bea839871a47f3fea7.png
[ca5d2f9eef141e086b564a767613107e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/adfe1c1b76004700946a6644f134f360.png
[c7bc92b369162562cf9e5c4c7c9d8f58.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/4aea2ca8487d442ca99a9f2b1d6cf5d1.png
[46cfbb687a89adf7ec0af8ac69fb43f4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/6473f9b0b11d4d629d9e4b82d2387763.png
[42111e2c22abfb39a3d49c0c0f85224f.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/88b71f6843ad465d9508ea5fa2c8e941.png
[accf03c573f5e8d5cd48aa97d8ae808c.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/2b080f5c5dcb493580753a8a316ea3a5.png
[cad0fd812bb0f92c479eb011021f88d0.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/6571c285b3f74e1a87f953866c8b38b2.png
[3b8fdb199f3a66ba077910d93e5266d7.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/7b4374adf66e44bfbe42039c1f0838af.png
[d0971a0d08b97a0f893b175cc825b23e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/812c0366791d4d349b946276e988ba99.png
[c5a0b63b5c5b10df961f57591d77240d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/093302fe319b4320a08533a1ac649aaf.png
[67ec50fedd944272403064f137488a4d.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/37092fa1951f4155bc79b3312d907b8e.png
[90043544b314add9685712f00b44d58f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/a77039c1d1c741899373da01f767b990.png
[cef97ff66956c63d78c8599bd84efc04.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/91fab9727db3462c9ee9e0d2bb20ad76.png