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

简介

推拉模式简介

一般而言我们在谈论推拉模式的时候指的是 Comsumer 和 Broker 之间的交互。

Producer 与 Broker 之间都是推的方式，即 Producer 将消息推送给 Broker，而不是 Broker 主动去拉取消息。原因如下：

如果需要 Broker 去拉取消息，那么 Producer 就必须在本地通过日志的形式保存消息来等待 Broker 的拉取，如果有很多生产者的话，那么消息的可靠性不仅仅靠 Broker 自身，还需要靠成百上千的 Producer。
Broker 还能靠多副本等机制来保证消息的存储可靠，而成百上千的 Producer 可靠性就有点难办了，所以默认的 Producer 都是推消息给 Broker。

各个MQ的推拉模式

MQ类别	推拉模式
RabbitMQ	推拉都支持。官方推荐使用推模式
RocketMQ	推拉都支持。（本质上，推模式也是拉模式）
Kafka	只有拉模式

推模式

简介

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

优点

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

缺点

推送速率难以适应消费速率

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

使用场景

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

拉模式

简介

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

优点

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

缺点

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

长轮询

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

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

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

RocketMQ

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

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

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

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

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

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

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

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

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

Kafka

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

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

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

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

先来看消费者端的代码。

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

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

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

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

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

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

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

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

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

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

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

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

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

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

其他网址

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

MQ--推模式与拉模式相关推荐

Media Player Classic - HC 源代码分析 14：PIN连接过程中推模式和拉模式区别
前面有两篇文章讲解了PIN连接过程中需要做2件事情: 媒体类型的协商分配器的协商推模式和拉模式关于媒体类型的协商调用的接口都一样,主要的区别还是在分配器的协商. 前面讲解的分配器的协商过程是推模式 ...
微博feed系统的推(push)模式和拉(pull)模式和时间分区拉模式架构探讨
sns系统,微博系统都应用到了feed(每条微博或者sns里的新鲜事等我们称作feed)系统,不管是twitter.com或者国内的新浪微博,人人网等,在各种技术社区,技术大会上都在分享自己的feed ...
【转发】微博feed系统的推(push)模式和拉(pull)模式和时间分区拉模式架构探讨
本文为转发文章,原文作者:孙立 sns系统,微博系统都应用到了feed(每条微博或者sns里的新鲜事等我们称作feed)系统,不管是twitter.com或者国内的新浪微博,人人网等,在各种技术社区, ...
RocketMQ消息消费方式推拉模式
RocketMQ消息消费本质上是基于的拉(pull)模式,consumer主动向消息服务器broker拉取消息. consumer被分为2类:MQPullConsumer和MQPushConsumer ...
MS6212可替换TI PCA9306、SGM4553用于开漏模式和推拉模式的 2bit 双向电平转换器
工采网代理的MS6212是一款双向电平转换器,开漏模式和推拉模式可以用作混合电压的数字信号系统中.其使用两个独立构架的电源供电,A端供电电压范围是1.65V到5.5V,B端供电电压范围是2.3V到5. ...
Alibaba Sentinel规则持久化-拉模式-手把手教程【基于文件】
文章目录一.拉模式架构二.原理简述三.编写 3.1 加依赖 3.2 写代码 3.3 配置四.优缺点分析五.你可能会有的疑问六.参考文档七.案例测试 7.1. 添加流控规则 7.2. 服务 ...
C语言单片机数码管a段亮,宏晶单片机设置强上拉模式让数码管更亮！
STC(宏晶)系列单片机的I/O口具有强上拉模式,何为强上拉模式? 宏晶单片机I/O口可以设置成四种模式:准双向口/弱上接,推挽/强上拉,输入/高阻,设置成推挽/强上拉,无论输出1和0时能承受的电流都 ...
关于MUI框架中，“侧滑导航“之“div模式下拉菜单“的a标签（超链接）的失效问题？
关于MUI框架中,"侧滑导航"之"div模式下拉菜单"的a标签(超链接)的失效问题? 截图和讲解图片的顺序如下所示: 图1 图2 · 官方的截图图3 · 官方 ...
AD10问题[拖动元器件时，无法移动至与之相邻元器件][3D报错Action not avaliable in 3D view][任意角度模式下拉不了任意角度的线]
拖动元器件时,无法移动至与之相邻元器件这个问题打开其他原理图也有,被逼得我都要重新装软件了!最后找到解决办法了! 本文摘录于:AD画PCB时,出现"拖动元器件时,无法移动至与之相邻元器件& ...
分布式事务——分布式事务简介、分布式事务框架 Seata（AT模式、Tcc模式、Tcc Vs AT）、分布式事务—MQ
分布式事务--分布式事务简介.分布式事务框架 Seata(AT模式.Tcc模式.Tcc Vs AT).分布式事务--MQ 一.分布式事务简介如果不是分布式环境的话一般不会接触到这种,一旦是微服务这种 ...

MQ--推模式与拉模式

简介

推模式

拉模式

长轮询

RocketMQ

Kafka

其他网址

MQ--推模式与拉模式相关推荐

最新文章

热门文章