MQ--推模式与拉模式

港控/mmm° 2022-10-23 15:27 150阅读 0赞

原文网址：[MQ--推模式与拉模式\_IT利刃出鞘的博客-CSDN博客][MQ--_IT_-CSDN]

# 其他网址 #

[消息队列之推还是拉，RocketMQ 和 Kafka 是如何做的？ - 云+社区 - 腾讯云][RocketMQ _ Kafka _ - _ -]

# 简介 #

**推拉模式简介**

> 一般而言我们在谈论**推拉模式的时候指的是 Comsumer 和 Broker 之间的交互。**
> 
> Producer 与 Broker 之间都是推的方式，即 Producer 将消息推送给 Broker，而不是 Broker 主动去拉取消息。原因如下：
> 
> 1.  如果需要 Broker 去拉取消息，那么 Producer 就必须在本地通过日志的形式保存消息来等待 Broker 的拉取，如果有很多生产者的话，那么消息的可靠性不仅仅靠 Broker 自身，还需要靠成百上千的 Producer。
> 2.  Broker 还能靠多副本等机制来保证消息的存储可靠，而成百上千的 Producer 可靠性就有点难办了，所以默认的 Producer 都是推消息给 Broker。

**各个MQ的推拉模式**

> <table style="width:500px;"> 
>  <tbody> 
>   <tr> 
>    <td style="width:136px;"><strong>MQ类别</strong></td> 
>    <td style="width:363px;"><strong>推拉模式</strong></td> 
>   </tr> 
>   <tr> 
>    <td style="width:136px;">RabbitMQ</td> 
>    <td style="width:363px;">推拉都支持。官方推荐使用推模式</td> 
>   </tr> 
>   <tr> 
>    <td style="width:136px;">RocketMQ</td> 
>    <td style="width:363px;">推拉都支持。（本质上，推模式也是拉模式）</td> 
>   </tr> 
>   <tr> 
>    <td style="width:136px;">Kafka</td> 
>    <td style="width:363px;">只有拉模式</td> 
>   </tr> 
>  </tbody> 
> </table>

# 推模式 #

**简介**

> 本处推模式指的是消息从 Broker 推向 Consumer，即 Consumer 被动的接收消息，由 Broker 来主导消息的发送。

**优点**

> 1.  **消息实时性高。**Broker 接受完消息之后可以立马推送给 Consumer。
> 2.  **对于消费者使用来说更简单。**消息来了就消费即可。

**缺点**

> **推送速率难以适应消费速率**
> 
> 1.  推模式的目标就是以最快的速度推送消息，当生产者往 Broker 发送消息的速率大于消费者消费消息的速率时，随着时间的增长消费者那边可能就“爆仓”了，因为根本消费不过来啊。当推送速率过快就像 DDos 攻击一样消费者就傻了。
> 2.  不同的消费者的消费速率还不一样，身为 Broker 很难平衡每个消费者的推送速率，如果要实现自适应的推送速率那就需要在推送的时候消费者告诉 Broker ，我不行了你推慢点吧，然后 Broker 需要维护每个消费者的状态进行推送速率的变更。  
>     这其实就增加了 Broker 自身的复杂度。

**使用场景**

> 推模式难以根据消费者的状态控制推送速率，适用于消息量不大、消费能力强要求实时性高的情况下。

# 拉模式 #

**简介**

> 本处拉模式指的是 Consumer 主动向 Broker 请求拉取消息，即 Broker 被动的发送消息给 Consumer。

**优点**

> 1.  **消费者可以根据自身的情况来发起拉取消息的请求**。假设当前消费者觉得自己消费不过来了，它可以根据一定的策略停止拉取，或者间隔拉取都行。
> 2.  **Broker 相对轻松了。**它只管存生产者发来的消息，至于消费的时候自然由消费者主动发起，来一个请求就给它消息呗，从哪开始拿消息，拿多少消费者都告诉它，它就是一个没有感情的工具人，消费者要是没来取也不关它的事。
> 3.  **可以更合适的进行消息的批量发送**，基于推模式可以来一个消息就推送，也可以缓存一些消息之后再推送，但是推送的时候其实不知道消费者到底能不能一次性处理这么多消息。而拉模式就更加合理，它可以参考消费者请求的信息来决定缓存多少消息之后批量发送。

**缺点**

> 1.  **消息延迟**，毕竟是消费者去拉取消息，但是消费者怎么知道消息到了呢？所以它只能不断地拉取，但是又不能很频繁地请求，太频繁了就变成消费者在攻击 Broker 了。因此需要降低请求的频率，比如隔个 2 秒请求一次，你看着消息就很有可能延迟 2 秒了。
> 2.  **消息忙请求**，忙请求就是比如消息隔了几个小时才有，那么在几个小时之内消费者的请求都是无效的，在做无用功。

# 长轮询 #

RocketMQ 和 Kafka 都是采用“长轮询”的机制来实现拉模式，具体的做法都是通过消费者等待消息，当有消息的时候 Broker 会直接返回消息，如果没有消息都会采取延迟处理的策略，并且为了保证消息的及时性，在对应队列或者分区有新消息到来的时候都会提醒消息来了，及时返回消息。

一句话说就是消费者和 Broker 相互配合，拉取消息请求不满足条件的时候 hold 住，避免了多次频繁的拉取动作，当消息一到就提醒返回。

为了简单化，下面我把消息不满足本次拉取的条数啊、总大小啊等等都统一描述成还没有消息，反正都是不满足条件。

## RocketMQ ##

RocketMQ 中的 PushConsumer 其实是披着**推**模式实际上是**拉**模式的方法，**只是看起来像推模式而已**。因为 RocketMQ 在被背后偷偷的帮我们去 Broker 请求数据了。

后台会有个 RebalanceService 线程，这个线程会根据 topic 的队列数量和当前消费组的消费者个数做负载均衡，每个队列产生的 pullRequest 放入阻塞队列 pullRequestQueue 中。然后又有个 PullMessageService 线程不断的从阻塞队列 pullRequestQueue 中获取 pullRequest，然后通过网络请求 broker，这样实现的准实时拉取消息。这一部分代码我不截了，就是这么个事儿，稍后会用图来展示。

然后 Broker 的 PullMessageProcessor 里面的 processRequest 方法是用来处理拉消息请求的，有消息就直接返回，如果没有消息怎么办呢？我们来看一下代码。

![373696d08b4bac7741a858c3513065a5.png][]

我们再来看下 suspendPullRequest 方法做了什么。

![7f289d1c8b2d0ab7dd4f9a55e50739a2.png][]

而 PullRequestHoldService 这个线程会每 5 秒从 pullRequestTable 取PullRequest请求，然后看看待拉取消息请求的偏移量是否小于当前消费队列最大偏移量，如果条件成立则说明有新消息了，则会调用 notifyMessageArriving ，最终调用 PullMessageProcessor 的 executeRequestWhenWakeup() 方法重新尝试处理这个消息的请求，也就是再来一次，整个长轮询的时间默认 30 秒。

![739fc86d2455e021144e0a00234b5318.png][]

简单的说就是 5 秒会检查一次消息时候到了，如果到了则调用 processRequest 再处理一次。这好像不太实时啊？ 5秒？

别急，还有个 ReputMessageService 线程，这个线程用来不断地从 commitLog 中解析数据并分发请求，构建出 ConsumeQueue 和 IndexFile 两种类型的数据，**并且也会有唤醒请求的操作，来弥补每 5s 一次这么慢的延迟**

代码我就不截了，就是消息写入并且会调用 pullRequestHoldService\#notifyMessageArriving。

最后我再来画个图，描述一下整个流程。

![66911b917af2234bb24891955440ad0d.png][]

## Kafka ##

像 Kafka 在拉请求中有参数，可以使得消费者请求在 “长轮询” 中阻塞等待。

简单的说就是消费者去 Broker 拉消息，定义了一个超时时间，也就是说消费者去请求消息，如果有的话马上返回消息，如果没有的话消费者等着直到超时，然后再次发起拉消息请求。

并且 Broker 也得配合，如果消费者请求过来，有消息肯定马上返回，没有消息那就建立一个延迟操作，等条件满足了再返回。

我们来简单的看一下源码，为了突出重点，我会删减一些代码。

先来看消费者端的代码。

![3a2163b03622a778f23ff37b687e95e4.png][]

上面那个 poll 接口想必大家都很熟悉，其实从注解直接就知道了确实是等待数据的到来或者超时，我们再简单的往下看。

![4e5216080606f50973b209319ffbe2ef.png][]

我们再来看下最终 client.poll 调用的是什么。

![7c146104e4713498bbc6a2b515229853.png][]

最后**调用的就是 Kafka 包装过的 selector，而最终会调用 Java nio 的 select(timeout)**。

现在消费者端的代码已经清晰了，**我们再来看看 Broker 如何做的**。

Broker 处理所有请求的入口其实我在之前的文章介绍过，就在 KafkaApis.scala 文件的 handle 方法下，这次的主角就是 handleFetchRequest 。

![92ba31262917a903c1734b508ff2d490.png][]

这个方法进来，我截取最重要的部分。

![62f36ce8e65b7cf40b7790853e02bec4.png][]

下面的图片就是 fetchMessages 方法内部实现，源码给的注释已经很清晰了，大家放大图片看下即可。

![03a13333ec6e53dae66e3f737ef46dc2.png][]

![62f36ce8e65b7cf40b7790853e02bec4.png][]

这个炼狱名字取得很有趣，简单的说就是利用我之前文章提到的时间轮，来执行定时任务，例如这里是`delayedFetchPurgatory`，专门用来处理延迟拉取操作。

我们先简单想一下，这个延迟操作都需要实现哪些方法，首先构建的延迟操作需要有检查机制，来查看消息是否已经到了，然后呢还得有个消息到了之后该执行的方法，还需要有执行完毕之后该干啥的方法，当然还得有个超时之后得干啥的方法。

这几个方法其实对应的就是代码里的 DelayedFetch ，这个类继承了 DelayedOperation 内部有：

*  isCompleted 检查条件是否满足的方法
 *  tryComplete 条件满足之后执行的方法
 *  onComplete 执行完毕之后调用的方法
 *  onExpiration 过期之后需要执行的方法

判断是否过期就是由时间轮来推动判断的，但是总不能等过期的时候再去看消息到了没吧？

这里 Kafka 和 RocketMQ 的机制一样，也会在消息写入的时候提醒这些延迟请求消息来了，具体代码我不贴了， 在 ReplicaManager\#appendRecords 方法内部再深入个两方法可以看到。

不过虽说代码不贴，图还是要画一下的。

![bca88f288247dc975cded17274f7257b.png][]

[MQ--_IT_-CSDN]: https://knife.blog.csdn.net/article/details/114291078
[RocketMQ _ Kafka _ - _ -]: https://cloud.tencent.com/developer/news/684791
[373696d08b4bac7741a858c3513065a5.png]: /images/20221023/d8057afa9e094507b77483789087e59f.png
[7f289d1c8b2d0ab7dd4f9a55e50739a2.png]: /images/20221023/436e9831668847b2bb42e4f92cf0c4c8.png
[739fc86d2455e021144e0a00234b5318.png]: /images/20221023/d9340e8fb4754ae083196382602b1daf.png
[66911b917af2234bb24891955440ad0d.png]: /images/20221023/372c7d985ff843939d389746036aae04.png
[3a2163b03622a778f23ff37b687e95e4.png]: /images/20221023/3a45ac8985e5472888d60f0c6e9d987b.png
[4e5216080606f50973b209319ffbe2ef.png]: /images/20221023/dda21ed8d9c54457852272274d7bb0a8.png
[7c146104e4713498bbc6a2b515229853.png]: /images/20221023/2af5521700724ef59e1b51fd0073e55c.png
[92ba31262917a903c1734b508ff2d490.png]: /images/20221023/b130a8c3652b47fda184025aa1daaa46.png
[62f36ce8e65b7cf40b7790853e02bec4.png]: /images/20221023/f2886451e7f942ca9c970761d99bed61.png
[03a13333ec6e53dae66e3f737ef46dc2.png]: /images/20221023/e7fb6424869b44668ad976419fb73c10.png
[bca88f288247dc975cded17274f7257b.png]: /images/20221023/2cf6ad533ef34e99804842ab594fcb9a.png