RocketMQ Consumer消息消费过程消息的拉取

Posted 2022-02-10 乐观男孩

目录

• 说明
• PullMessageService
• 消息拉取流程图
• 消费拉取过程
• 总结

说明

从上一节《Consumer消息消费过程（一）、queue均衡》中，我们知道在Consumer启动的过程中，会触发queue均衡。queue均衡主要完成以下几个工作：
1、按照一定的算法对Topic内的queue进行均衡，均衡过后就明确了当前Consumer该消费哪些queue；
2、每个需要消费的queue就会封装成一个消息拉取任务PullRequest，交由PullMessageService异步执行消息拉取任务。
在本节中，我们就学习下消息的拉取过程。

PullMessageService

PullMessageService，是消息拉取的入口，该类维护了一个无界的阻塞队列pullRequestQueue和一个线程池scheduledExecutorService。阻塞队列存放的是待进行消息拉取的任务。线程池负责将消息拉取任务放入队列（线程池并不具体执行消息拉取任务，消息拉取任务在run方法中进行）。

PullMessageService实现了Runnable接口，并且在MQClientInstance启动的过程中就启动了PullMessageService。PullMessageService会一直执行run方法，从队列中取出消息拉取任务，执行消息的拉取。

消息拉取流程图

消费拉取过程

1、PullMessageService.pullMessage，实际将消息拉取任务委托给DefaultMQPushConsumerImpl

2、DefaultMQPushConsumerImpl.pullMessage()方法，主要完成以下的工作：
（1）、重新计算消费的offset

默认计算offset的策略是CONSUME_FROM_LAST_OFFSET（从最新的offset开始消费）。
首先从offsetStore获取，获取不到（代表是第一次消费）时：如果是重试消费Topic，则直接从头开始消费，否则从Broker中获取queue最大的offset，从该offset进行消费。

（2）、定义消息消费的回调方法

（3）、构造是否需要提交offset、是否需要过滤消息等标识位

（4）、调用pullAPIWrapper提交消息拉取请求（消息拉取为异步方式：CommunicationMode.ASYNC）

3、PullAPIWrapper.pullKernelImpl()方法，主要完成以下的工作：
（1）、查找Broker地址

（2）、构造PullMessageRequestHeader，调用MQClientAPIImpl接口进行消息拉取

4、MQClientAPIImpl.pullMessage()直接调用pullMessageAsync()方法

通过NettyRemotingClient异步请求Broker，拉取消息。拉取成功后，调用DefaultMQPushConsumerImpl定义的回调函数进行消息的处理。

总结

1、Consumer默认是从最新的offset进行消费（CONSUME_FROM_LAST_OFFSET）；
2、消息拉取过程中，是采用异步的方式进行处理，Broker返回结果时通过调用回调方法的方式进行结果处理。所以，当队列中有消息拉取任务时，能够很快的被处理，不会被阻塞。