rocketmq--push消费过程

谁践踏了优雅 2022-05-04 03:58 240阅读 0赞

转载: [https://www.cnblogs.com/chenjunjie12321/p/7922362.html][https_www.cnblogs.com_chenjunjie12321_p_7922362.html]

**Rocketmq消费分为push和pull两种方式，push为被动消费类型，pull为主动消费类型，push方式最终还是会从broker中pull消息。不同于pull的是，push首先要注册消费监听器，当监听器处触发后才开始消费消息，所以被称为“被动”消费。**

具体地，以pushConsumer的测试例子展开介绍，通常使用push消费的过程如下：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 </td> 
 <td> <code>public</code>&nbsp;<code>class</code>&nbsp;<code>PushConsumer { </code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>public</code>&nbsp;<code>static</code>&nbsp;<code>void</code>&nbsp;<code>main(String[] args)&nbsp;</code><code>throws</code>&nbsp;<code>InterruptedException, MQClientException { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>DefaultMQPushConsumer consumer =&nbsp;</code><code>new</code>&nbsp;<code>DefaultMQPushConsumer(</code><code>"CID_JODIE_1"</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>consumer.subscribe(</code><code>"Jodie_topic_1023"</code><code>,&nbsp;</code><code>"*"</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//wrong time format 2017_0422_221800</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>consumer.setConsumeTimestamp(</code><code>"20170422221800"</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>consumer.registerMessageListener(</code><code>new</code>&nbsp;<code>MessageListenerConcurrently() { </code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>@Override</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>public</code>&nbsp;<code>ConsumeConcurrentlyStatus consumeMessage(List&lt;MessageExt&gt; msgs, ConsumeConcurrentlyContext context) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>System.out.printf(Thread.currentThread().getName() +&nbsp;</code><code>" Receive New Messages: "</code>&nbsp;<code>+ msgs +&nbsp;</code><code>"%n"</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code>&nbsp;<code>ConsumeConcurrentlyStatus.CONSUME_SUCCESS;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>});</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>consumer.start();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>System.out.printf(</code><code>"Consumer Started.%n"</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

上述过程背后设计到的点如下：

**I. checkConfig 检查内容：**

1.消费组 -- （不能与默认DEFAULT\_CONSUMER同名）

2.消费模型 -- （默认CLUSTERING）

3.从何处开始消费 -- （默认CONSUME\_FROM\_LAST\_OFFSET）

4.消费时间戳 -- （消息回溯，默认Default backtracking consumption time Half an hour ago）

5.消费负载均衡策略 -- （默认AllocateMessageQueueAveragely）

6.订阅关系 --（map类型，即可订阅多个topic；key=Topic, value=订阅描述）

7.消费监听 --（必须为orderly or concurrently类型之一）

8.消费消息的线程数量控制 -- （消费线程池最大、最小数量）

9.检查单队列并行消费允许的最大跨度 --（consumeConcurrentlyMaxSpan）

10.检查拉消息本地队列缓存消息最大数 --（pullThresholdForQueue）（processQueue.getMsgCount()记数）

11.检查拉取时间间隔 --（拉消息间隔，由于是长轮询，所以默认为0）

12.检查批量消费的个数 --（一次消费多少条消息）

13.检查批量拉取消息的个数 --（一次最多拉多少条）

**II. copySubscription：**

将订阅信息设置到rebalanceImpl的map中用于负载。另外，如果该消费者的消费模式为集群消费，则会将retry的topic一并放到rebalanceImpl的map中用于负载。

**III. 设置rebanlance信息**

**IV. 实例化pull消息的包装类型**

**V. 如果不存在offsetStore对象，实例化offsetStore**

广播模式：

public class LocalFileOffsetStore implements OffsetStore \{...\}

注：load()函数体不为空

集群模式：

public class RemoteBrokerOffsetStore implements OffsetStore \{...\}

注：load()函数体为空

**VI. 获取监听器，实例化consumeMessageService服务并启动**

ConsumeMessageOrderlyService启动后会对拉取下来的消息进行处理。ConsumeMessageOrderlyService有两种类型：ConsumeMessageOrderlyService和ConsumeMessageConcurrentlyService。

1). 如果消息监听器是orderly类型，则创建ConsumeMessageOrderlyService实例

ConsumeMessageOrderlyService.start()只处理消息模式为CLUSTERING的消息消费。

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 </td> 
 <td> <code>public</code>&nbsp;<code>void</code>&nbsp;<code>start() { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code>&nbsp;<code>(MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.</code><code>this</code><code>.defaultMQPushConsumerImpl</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>.messageModel())) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.scheduledExecutorService.scheduleAtFixedRate(</code><code>new</code>&nbsp;<code>Runnable() { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>@Override</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>public</code>&nbsp;<code>void</code>&nbsp;<code>run() { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>ConsumeMessageOrderlyService.</code><code>this</code><code>.lockMQPeriodically();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>},&nbsp;</code><code>1000</code>&nbsp;<code>*&nbsp;</code><code>1</code><code>, ProcessQueue.RebalanceLockInterval, TimeUnit.MILLISECONDS);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

线程启动后会每隔20s执行lockMQPeriodicallys()，lockMQPeriodicallys()会将消费的队列上锁，然后处理，具体过程，有机会单独成文分析。

2). 如果消息监听器是concurrently类型，则创建ConsumeMessageConcurrentlyService实例

ConsumeMessageConcurrentlyService.start()会定时清除过期消息 --> cleanExpireMsg()。

**VII. 注册消费组**

将group和consumer注册到MQClientInstance实例。

与生产者注册生产者组类似，一个客户端进程中一个consumerGroup只能有一个实例。

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 </td> 
 <td> <code>MQConsumerInner prev =&nbsp;</code><code>this</code><code>.consumerTable.putIfAbsent(group, consumer);</code> <code>if</code>&nbsp;<code>(prev !=&nbsp;</code><code>null</code><code>) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>log.warn(</code><code>"the consumer group["</code>&nbsp;<code>+ group +&nbsp;</code><code>"] exist already."</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code>&nbsp;<code>false</code><code>;</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

如果没有注册成功，则关闭消费服务，consumeMessageService.shutdown()。

**VIII. 启动mQClientFactory及MQClientInstance**

1). 获取client实例对象MQClientInstance -- getAndCreateMQClientInstance。一个进程只能产生一个MQClientInstance实例对象, 某个客户端的生产者与消费者共用这个实例对象。

2). 启动客户端实例的个各种服务：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 </td> 
 <td> <code>public</code>&nbsp;<code>void</code>&nbsp;<code>start()&nbsp;</code><code>throws</code>&nbsp;<code>MQClientException { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>synchronized</code>&nbsp;<code>(</code><code>this</code><code>) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>switch</code>&nbsp;<code>(</code><code>this</code><code>.serviceState) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>case</code>&nbsp;<code>CREATE_JUST:</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.serviceState = ServiceState.START_FAILED;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 1.判断NamesrvAddr是否为空，为空去远程http服务拉去地址</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code>&nbsp;<code>(</code><code>null</code>&nbsp;<code>==&nbsp;</code><code>this</code><code>.clientConfig.getNamesrvAddr()) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.clientConfig.setNamesrvAddr(</code><code>this</code><code>.mQClientAPIImpl.fetchNameServerAddr());</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 2.开启通信服务</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.mQClientAPIImpl.start();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 3.启动各种定时任务</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.startScheduledTask();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 4.启动消息拉取服务，循环拉取阻塞队列pullRequestQueue</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.pullMessageService.start();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 5. 启动负载均衡服务</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.rebalanceService.start();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// 6.启动消息生产服务</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.defaultMQProducer.getDefaultMQProducerImpl().start(</code><code>false</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>log.info(</code><code>"the client factory [{}] start OK"</code><code>,&nbsp;</code><code>this</code><code>.clientId);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.serviceState = ServiceState.RUNNING;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>break</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>case</code>&nbsp;<code>RUNNING:</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>break</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>case</code>&nbsp;<code>SHUTDOWN_ALREADY:</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>break</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>case</code>&nbsp;<code>START_FAILED:</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>throw</code>&nbsp;<code>new</code>&nbsp;<code>MQClientException(</code><code>"The Factory object["</code>&nbsp;<code>+&nbsp;</code><code>this</code><code>.getClientId()</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>+&nbsp;</code><code>"] has been created before, and failed."</code><code>,&nbsp;</code><code>null</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>default</code><code>:</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>break</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

分析push消费的过程，需对上述过程第3点、第4点、第5点依次介绍。

第3点、启动各种定时任务过程：

<table> 
 <tbody> 
 <tr> 
 <td>&nbsp;编号&nbsp;</td> 
 <td>任务</td> 
 <td>周期</td> 
 <td>启动时延&nbsp;</td> 
 </tr> 
 <tr> 
 <td>1&nbsp;</td> 
 <td>&nbsp;获取namesrv地址</td> 
 <td>每隔2分钟&nbsp;</td> 
 <td>0.01s</td> 
 </tr> 
 <tr> 
 <td>2</td> 
 <td>&nbsp;更新路由信息</td> 
 <td>每隔3分钟&nbsp;</td> 
 <td>001s</td> 
 </tr> 
 <tr> 
 <td>3</td> 
 <td>&nbsp;向所有broker发送心跳包，并清除无效broker &nbsp;</td> 
 <td>每隔30s&nbsp;</td> 
 <td>1s</td> 
 </tr> 
 <tr> 
 <td>4</td> 
 <td>&nbsp;持久化消费位置offset</td> 
 <td>每隔5s&nbsp;</td> 
 <td>10s</td> 
 </tr> 
 <tr> 
 <td>5</td> 
 <td>&nbsp;调整消费线程池大小</td> 
 <td>每隔1分钟&nbsp;</td> 
 <td>1min</td> 
 </tr> 
 </tbody> 
</table>

注：编号3中，客户端会通过心跳消息，向broker注册消费信息。Broker收到该心跳消息，把它维护在一个叫做ConsumerManager的对象里面，为之后做消费的负载均衡提供数据，负载均衡在消费端做，消费端在负载均衡时首先要从broker那获取这份全局信息。

第4点 启动pullMessageService服务

初始化客户端实例时，创建PullMessageService服务对象。

this.pullMessageService = new PullMessageService(this)，其中PullMessageService继承于ServiceThread，是一个线程对象。启动消息拉取服务线程后，在线程没有阻塞的情况下会不断地从循环阻塞队列pullRequestQueue拉取PullRequest对象，然后执行this.pullMessage(pullRequest)。

那么pullRequestQueue的数据如何put进去的?核心是doRebalance ，负载均衡具体细节可以参考：[http://www.cnblogs.com/chenjunjie12321/p/7913323.html][http_www.cnblogs.com_chenjunjie12321_p_7913323.html]。

![1072097-20180313222110715-1434247200.png][]

例如当前有N个客户端同时消费一个topic下的消息队列（如上图），当前客户端( clientId = currentCId)，经过负载均衡处理后得到分配给当前消费者的消息队列（如上图的qM、qN），之后将这些队列与processQueueTable中的队列进行比对分析，见下面第五点。

第5点 RebalancePushImpl 负载均衡，分发pullRequest到pullRequestQueue。

负载均衡处理后得到分配给当前消费者的消息队列，然后将这些队列进行updateProcessQueueTableInRebalance 处理。updateProcessQueueTableInRebalance 的大致逻辑为如下 I、II 两步：

![1072097-20171129193926651-119860353.png][]

I. 首先检查当前RebalancePushImpl实例processQueueTable中与mqSet的包含关系

（1）如图中processQueueTable的灰色部分，表示与mqSet集合不互不包含的队列，对这些队列首先设置Dropped为true，然后看这些队列是否可以移除出processQueueTable--removeUnnecessaryMessageQueue，即每隔1s 看是否可以拿到当前队列的消费锁（tryLock()），拿到后返回true, 如果等待1s后仍然拿不到当前队列的消费锁则返回false，如果返回true则从processQueueTable移除对应的Entry<MessageQueue, ProcessQueue>；

（2） 如图中processQueueTable的白色部分，表示与mqSet集合的交集队列，对于这些队列，如果是消费类型是pull型，则不用管，如果是push型，看这些队列是否isPullExpired，如果是这些队列首先设置Dropped为true，则可以移除出processQueueTable--removeUnnecessaryMessageQueue。

II. 经过 I 处理，processQueueTable更新之后， 将processQueueTable集合与mqSet的的相对补集: processQueueTable(mq) - mqSet 里的消息队列依次封装成pullRequest，然后dispatchPullRequest到pullRequestQueue中。

经过上述处理后，待消费的队列放在了pullRequestList中，之后遍历pullRequestList，对遍历的每个队列进行消费，代码如下：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 </td> 
 <td> <code>@Override</code> <code>&nbsp;&nbsp;&nbsp;</code><code>public</code>&nbsp;<code>void</code>&nbsp;<code>dispatchPullRequest(List&lt;PullRequest&gt; pullRequestList) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>for</code>&nbsp;<code>(PullRequest pullRequest : pullRequestList) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.defaultMQPushConsumerImpl.executePullRequestImmediately(pullRequest);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>log.info(</code><code>"doRebalance, {}, add a new pull request {}"</code><code>, consumerGroup, pullRequest);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

executePullRequestImmediately的逻辑功能：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 </td> 
 <td> <code>public</code>&nbsp;<code>void</code>&nbsp;<code>executePullRequestImmediately(</code><code>final</code>&nbsp;<code>PullRequest pullRequest) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>try</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.pullRequestQueue.put(pullRequest);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}&nbsp;</code><code>catch</code>&nbsp;<code>(InterruptedException e) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>log.error(</code><code>"executePullRequestImmediately pullRequestQueue.put"</code><code>, e);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

总之，最终会将负载均衡得到的队列存放到pullRequestQueue。

回过来继续分析第4点，pullMessageService线程涉及到消费的核心过程DefaultMQPushConsumerImpl.pullMessage，pullMessageService线程线程体源码如下：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 </td> 
 <td> <code>@Override</code> <code>&nbsp;&nbsp;&nbsp;</code><code>public</code>&nbsp;<code>void</code>&nbsp;<code>run() { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>log.info(</code><code>this</code><code>.getServiceName() +&nbsp;</code><code>" service started"</code><code>);</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>while</code>&nbsp;<code>(!</code><code>this</code><code>.isStopped()) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>try</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>PullRequest pullRequest =&nbsp;</code><code>this</code><code>.pullRequestQueue.take();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code>&nbsp;<code>(pullRequest !=&nbsp;</code><code>null</code><code>) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.pullMessage(pullRequest);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}&nbsp;</code><code>catch</code>&nbsp;<code>(InterruptedException e) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}&nbsp;</code><code>catch</code>&nbsp;<code>(Exception e) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>log.error(</code><code>"Pull Message Service Run Method exception"</code><code>, e);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>log.info(</code><code>this</code><code>.getServiceName() +&nbsp;</code><code>" service end"</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

调用DefaultMQPushConsumerImpl.pullMessage方法：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 </td> 
 <td> <code>private</code>&nbsp;<code>void</code>&nbsp;<code>pullMessage(</code><code>final</code>&nbsp;<code>PullRequest pullRequest) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>final</code>&nbsp;<code>MQConsumerInner consumer =&nbsp;</code><code>this</code><code>.mQClientFactory.selectConsumer(pullRequest.getConsumerGroup());</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code>&nbsp;<code>(consumer !=&nbsp;</code><code>null</code><code>) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>DefaultMQPushConsumerImpl impl = (DefaultMQPushConsumerImpl) consumer;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>impl.pullMessage(pullRequest);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}&nbsp;</code><code>else</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>log.warn(</code><code>"No matched consumer for the PullRequest {}, drop it"</code><code>, pullRequest);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}&nbsp;</code> </td> 
 </tr> 
 </tbody> 
</table>

pullMessage具体体拉流程如下图所示：

![1072097-20171129230734401-1473917554.png][]

下面对并发消费模型（concurrently）的消费代码进行展示：
    
    class ConsumeRequest implements Runnable ，其线程体方法如下：

![ContractedBlock.gif][]

consumerRequest逻辑：

![1072097-20180621222259299-1348507208.png][]

![1072097-20180621223129436-330573371.png][]

processConsumeResult -- 对消费结果进行处理：

![1072097-20180621233918659-2004797148.png][]

![1072097-20180621234256624-1072403545.png][]

重试队列发消息逻辑：

![1072097-20180621231906387-2080355172.png][]

生成一个重试队列，重试队列topic =  %RETRY% + consumerGroup的形式。

附：

值得注意的是每次消费pullRequest上的一条数据后上更新消费到达的 offset，然后将pullRequest.setNextOffset(offset);

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 </td> 
 <td> <code>//这里的 this 为一个 DefaultMQPushConsumerImpl 实例对象</code> <code>final</code>&nbsp;<code>long</code>&nbsp;<code>offset =&nbsp;</code><code>this</code><code>.rebalanceImpl.computePullFromWhere(pullRequest.getMessageQueue());</code> <code>...</code> <code>pullRequest.setNextOffset(offset);</code> </td> 
 </tr> 
 </tbody> 
</table>

其中 computePullFromWhere采用的策略有如下三种（另外还有几个已经被弃用的（@Deprecated））：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 </td> 
 <td> <code>CONSUME_FROM_LAST_OFFSET（默认）： 一个新的消费集群第一次启动从队列的最后位置开始消费。后续再启动接着上次消费的进度开始消费。</code> <code>CONSUME_FROM_FIRST_OFFSET： 一个新的消费集群第一次启动从队列的最前位置开始消费。后续再启动接着上次消费的进度开始消费。</code> <code>CONSUME_FROM_TIMESTAMP： 一个新的消费集群第一次启动从指定时间点开始消费。后续再启动接着上次消费的进度开始消费。</code> </td> 
 </tr> 
 </tbody> 
</table>

DefaultMQPushConsumer 中默认采用 CONSUME\_FROM\_LAST\_OFFSET 这种方式，当然可以根据自己需要修改computePullFromWhere的策略

<table> 
 <tbody> 
 <tr> 
 <td> 1 </td> 
 <td> <code>private</code>&nbsp;<code>ConsumeFromWhere consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET;</code> </td> 
 </tr> 
 </tbody> 
</table>

**IX. updateTopicSubscribeInfoWhenSubscriptionChanged**

**X. sendHeartbeatToAllBrokerWithLock**

**XI. rebalanceImmediately**

（完）

当你的才华还撑不起你的野心的时候，快来这里愉快的学习吧

wx公众号:java架构狮  干货满满  视频多多

![70][]

[https_www.cnblogs.com_chenjunjie12321_p_7922362.html]: https://www.cnblogs.com/chenjunjie12321/p/7922362.html
[http_www.cnblogs.com_chenjunjie12321_p_7913323.html]: http://www.cnblogs.com/chenjunjie12321/p/7913323.html
[1072097-20180313222110715-1434247200.png]: /images/20220504/3a8d99ec1a854ac69f186f905cd08648.png
[1072097-20171129193926651-119860353.png]: /images/20220504/6ca799fea98c4857b889d0d3426d3083.png
[1072097-20171129230734401-1473917554.png]: /images/20220504/28b373a7e5b041209682356676d9e8a5.png
[ContractedBlock.gif]: /images/20220504/e964a38dbb8a4525b4b6c0bb6f349bc9.png
[1072097-20180621222259299-1348507208.png]: /images/20220504/7ed6a6dc20d8442095c6c0bae31b665c.png
[1072097-20180621223129436-330573371.png]: /images/20220504/4b6e1ce8d9184ae68f19cc9b7a126054.png
[1072097-20180621233918659-2004797148.png]: /images/20220504/237b32a184ad42beb18572a9f6ab603a.png
[1072097-20180621234256624-1072403545.png]: /images/20220504/56ed70c4523849709cd1fae96e6d53f9.png
[1072097-20180621231906387-2080355172.png]: /images/20220504/d79118529cc345c6b6302a9ed396c696.png
[70]: /images/20220504/9605e6be739241a3897f789c6b9c5baa.png