RocketMQ实战-初探PushConsumer限流机制实现原理

往期回顾

基于V4.9.4版本的RocketMQ源码解析专栏

RocketMQ实战-对于消息生产者使用的一些思考

上一篇文章我们总结了对于RocketMQ消息生产者如何使用的一些思考。旨在帮助广大开发者用好消息生产者，提高生产环境的稳定性以及发送效率。

本篇文章，我们着重分析一下RocketMQ中PushConsumer的限流机制是如何实现的。在实现的过程中，客户端做了哪些事情，服务端又做了哪些事情，这将是本篇重点分析的地方。

PushConsumer消费模型

客户端实现消息拉取简易的模型，右侧是消息拉取涉及的类以及用数字标识的拉取流程，左侧代表最初的PullRequest是如何构建出来的。

我们了解到，在RocketMQ中，虽然消息消费被命名为了Push，但是本质上还是拉的模型，只不过这个拉取的动作被封装到了客户端API的内部，在使用者看来，消息是源源不断的被Push了过来。

消费端限流机制

PullMessageService这个线程类负责源源不断的拉取消息供客户端线程池消费，并且将拉取下来的消息存储在ProcessQueue中。

一个MessageQueue可以理解为一个服务端的队列，一个ProcessQueue可以理解为服务端的队列在客户端的虚拟镜像，ProcessQueue有两个属性和一个方法，跟客户端限流息息相关。

如果线程池消费的速度跟不上拉取的速度，并且不加以控制的话，消费者内部持有的线程池中的任务队列积攒的任务就会越来越多，容易造成内容溢出等异常情况，为了加以限制，保障客户端稳定的运行，RocketMQ客户端实现了三个维度的限流机制。

积压的消息个数限流

如果ProcessQueue中积压的消息个数大于1000个，则延迟拉取本次请求，给予消费者消费消息的时间。

积压的消息大小限流

如果ProcessQueue中积压的消息个数大于100M，则延迟拉取本次请求，给予消费者消费消息的时间。

偏移量大小限流

关于偏移量大小的限流，图示说明一下:

如果消费者在消费1-Msg时卡住了，但是消息拉取线程仍在源源不断的拉取消息，比如说拉取到了3000-Msg的消息，但是基于RocketMQ的位点提交机制，此时其他的消息消费完成仍然会提交1-Msg的消息的位点，这样的话，当消费者重启或者发生队列重平衡，将会导致1-Msg至3000-Msg之间的消息重新被消费，故引入了这条限流机制。

服务端限流机制

中间件服务端的限流，是中间件的一种自我保护的机制，为了保护自己的稳定运行，特别是在用户使用不当时。

服务端的限流源码实现在DefaultMessageStore的getMessage方法中，具体代码如下:

判断本地拉取的消息是在内存中还是在物理磁盘上面的方法。这是一种经验值的预估方法，一般认为待拉取的消息偏移量与最大的物理偏移量之间的差值超过了内存的40%，则会被判断为从磁盘拉取。

这种判断只是经验之谈，并不是十分准确。

不管是在内存中还是在磁盘上面，只要满足其中一条,就会触发限流。

若待拉取的消息在磁盘上面，不在内存中，并且一次性拉取的消息量较大，这种情况就会产生大量的缺页中断，导致大量的历史PageCahce页加载到内存中，也会进一步影响到消息的发送，为了解决这种问题，引入了上面的限流机制。

本篇总结

本篇从客户端以及服务端的视角，分别介绍了各自的限流机制，我们从中可以看出来，站在不同的视角，限流策略考虑的方向也是不同的，客户端主要是为了防止内存的溢出以及重复消费，服务端主要是为了保障中间件运行的稳定性。

希望本篇的分析能够帮助到大家。