老弟问我，RocketMQ中的ProcessQueue怎么理解？

　　今天来分享 RocketMQ 中一个非常重要又不太好理解的知识点-ProcessQueue。
　　一句话概括，ProcessQueue 就是 MessageQueue 的消费快照。看下面这张图：
　　1 ProcessQueue 构建
　　RocketMQ 客户端启动时，会开启一个 rebalance 线程，代码如下: //MQClientInstance.java public void start() throws MQClientException {  synchronized (this) {   switch (this.serviceState) {    case CREATE_JUST:     //...     // Start rebalance service     this.rebalanceService.start();    //...   }  } }
　　这个线程会不停的做重平衡操作，对 ProcessQueue 进行维护。在重平衡线程类 RebalanceImpl 定义了一个变量 processQueueTable，数据结构如下：
　　可以看到，在 processQueueTable 这个数据结构上维护了 MessageQueue 和 ProcessQueue 的映射。
　　下面看一下维护 processQueueTable 的代码： private boolean updateProcessQueueTableInRebalance(final String topic, final Set mqSet,  final boolean isOrder) {  boolean changed = false;   Iterator> it = this.processQueueTable.entrySet().iterator();  while (it.hasNext()) {   Entry next = it.next();   MessageQueue mq = next.getKey();   ProcessQueue pq = next.getValue();    if (mq.getTopic().equals(topic)) {    if (!mqSet.contains(mq)) {     //从processQueueTable上移除    } else if (pq.isPullExpired()) {     switch (this.consumeType()) {      case CONSUME_ACTIVELY://拉模式       break;      case CONSUME_PASSIVELY://推模式       //从processQueueTable上移除       break;      default:       break;     }    }   }  }     //创建ProcessQueue并放到processQueueTable  List pullRequestList = new ArrayList();  for (MessageQueue mq : mqSet) {   if (!this.processQueueTable.containsKey(mq)) {    //...    ProcessQueue pq = new ProcessQueue();     long nextOffset = -1L;    try {     nextOffset = this.computePullFromWhereWithException(mq);    } catch (Exception e) {     log.info(＂doRebalance, {}, compute offset failed, {}＂, consumerGroup, mq);     continue;    }     if (nextOffset >= 0) {     ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);     if (pre != null) {      log.info(＂doRebalance, {}, mq already exists, {}＂, consumerGroup, mq);     } else {         //封装好processQueueTable后再创建一个PullRequest进行消息拉取      log.info(＂doRebalance, {}, add a new mq, {}＂, consumerGroup, mq);      PullRequest pullRequest = new PullRequest();      pullRequest.setConsumerGroup(consumerGroup);      pullRequest.setNextOffset(nextOffset);      pullRequest.setMessageQueue(mq);      pullRequest.setProcessQueue(pq);      pullRequestList.add(pullRequest);      changed = true;     }    } else {     log.warn(＂doRebalance, {}, add new mq failed, {}＂, consumerGroup, mq);    }   }  }   this.dispatchPullRequest(pullRequestList);   return changed; } 2 拉取消息
　　上一节中构建 ProcessQueue 后，会再创建一个 PullRequest，这个 PullRequest 封装了 MessageQueue 和 ProcessQueue，创建成功后被放到了 PullMessageService 中的 pullRequestQueue 变量： //PullMessageService.java private final LinkedBlockingQueue pullRequestQueue = new LinkedBlockingQueue();  public void executePullRequestImmediately(final PullRequest pullRequest) {  try {   this.pullRequestQueue.put(pullRequest);  } catch (InterruptedException e) {   log.error(＂executePullRequestImmediately pullRequestQueue.put＂, e);  } }
　　这里以 RocketMQ 的推模式为例，Consumer 拉取到消息后，会进行如下处理： 对拉取到的消息根据 TAG 再次 进行过滤； 更新 PullRequest 下次拉取的偏移量 nextOffset；
　　把拉取的消息封装到 ProcessQueue 的 msgTreeMap（ 放到 msgTreeMap 之前首先要获取到写锁 treeMapLock ）； 封装 ConsumeRequest 进行消息消费; 封装消息拉取请求再次进行拉取。
　　代码如下： //DefaultMQPushConsumerImpl.java public void onSuccess(PullResult pullResult) {  if (pullResult != null) {      //1. 对拉取到的消息根据 TAG 再次进行过滤   pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,    subscriptionData);    switch (pullResult.getPullStatus()) {    case FOUND:     //2. 更新 PullRequest 下次拉取的偏移量 nextOffset     pullRequest.setNextOffset(pullResult.getNextBeginOffset());          if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {      DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);     } else {      //3. 把拉取的消息封装到 ProcessQueue 的 msgTreeMap      boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());      //4. 封装 ConsumeRequest 进行消息消费      DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(       pullResult.getMsgFoundList(),       processQueue,       pullRequest.getMessageQueue(),       dispatchToConsume);                     //5. 封装消息拉取请求      if (DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval() > 0) {       DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest,        DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval());      } else {       DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);      }     }     break;    //...   }  } } 3 消费消息
　　在上一节提到过，拉取到消息后，会把消息封装成一个 ConsumeRequest,这个线程类会调用消费者定义的 MessageListener 进行消费处理。看一下源代码： //ConsumeMessageConcurrentlyService.ConsumeRequest public void run() {  if (this.processQueue.isDropped()) {   log.info(＂the message queue not be able to consume, because it＂s dropped. group={} {}＂, ConsumeMessageConcurrentlyService.this.consumerGroup, this.messageQueue);   return;  }   MessageListenerConcurrently listener = ConsumeMessageConcurrentlyService.this.messageListener;  ConsumeConcurrentlyContext context = new ConsumeConcurrentlyContext(messageQueue);  ConsumeConcurrentlyStatus status = null;   try {   status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);  }//...   if (!processQueue.isDropped()) {   ConsumeMessageConcurrentlyService.this.processConsumeResult(status, context, this);  } }
　　消息消费成功后，会调用 processConsumeResult 方法进行结果处理。对于广播模式，发送失败后不会做重试，相当于把消息丢弃，而对于集群模式，消费失败的消息会发送到 Broker 端等待消费者重新拉取进行重试。
　　消费结果处理完后，消费成功的消息会从 ProcessQueue 的 msgTreeMap 中移除（ 需要获取到写锁 treeMapLock ），同时从 msgTreeMap 中获取最小的 Offset 来更新对应 MessageQueue 的偏移量。这个逻辑可以参考下面代码：public void processConsumeResult(  final ConsumeConcurrentlyStatus status,  final ConsumeConcurrentlyContext context,  final ConsumeRequest consumeRequest ) {  int ackIndex = context.getAckIndex();   switch (status) {   case CONSUME_SUCCESS:    if (ackIndex >= consumeRequest.getMsgs().size()) {     ackIndex = consumeRequest.getMsgs().size() - 1;    }    int ok = ackIndex + 1;    break;   //...  }  switch (this.defaultMQPushConsumer.getMessageModel()) {   case BROADCASTING:    //...    break;   case CLUSTERING:    List msgBackFailed = new ArrayList(consumeRequest.getMsgs().size());    for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {     MessageExt msg = consumeRequest.getMsgs().get(i);     //消费失败的，发送回Broker     boolean result = this.sendMessageBack(msg, context);     //...    }     break;   default:    break;  }     //从msgTreeMap中移除并返回msgTreeMap第一条消息的offset  long offset = consumeRequest.getProcessQueue().removeMessage(consumeRequest.getMsgs());  if (offset >= 0 && !consumeRequest.getProcessQueue().isDropped()) {   this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), offset, true);  } } 4 消费者限流4.1 缓存消息数量
　　如果消费者缓存的消息数量大于 RocketMQ 配置的阈值（默认 1000），就会触发延迟拉取，而消费者缓存的消息数量就来自 ProcessQueue，看下面代码： long cachedMessageCount = processQueue.getMsgCount().get(); if (cachedMessageCount > this.defaultMQPushConsumer.getPullThresholdForQueue()) {  this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);  return; } 4.2 缓存的消息大小
　　如果消费者缓存的消息大小大于 RocketMQ 配置的阈值（默认 100M），就会触发延迟拉取，而消费者缓存的消息大小就来自 ProcessQueue，看下面代码： long cachedMessageSizeInMiB = processQueue.getMsgSize().get() / (1024 * 1024); if (cachedMessageSizeInMiB > this.defaultMQPushConsumer.getPullThresholdSizeForQueue()) {  this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);  return; } 4.3 消息间隔
　　对于普通消息，如果消费偏移量间隔大于配置的阈值（默认 2000），就会触发延迟拉取，而消息间隔就来自 ProcessQueue，看下面代码： if (!this.consumeOrderly) {  if (processQueue.getMaxSpan() > this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan()) {   this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);   return;  } } 4.4 获取锁失败
　　对于顺序消息，如果获取锁失败，也会触发延迟拉取，而判断获取锁是否成功，也是在 ProcessQueue，看下面代码： if (processQueue.isLocked()) {  //... } else {  this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException); } 5 总结
　　ProcessQueue 是 MessageQueue 的消费快照，可以协助消费者进行消息拉取、消息消费、更新偏移量、限流。最后，看一下 ProcessQueue 的数据结构：
　　来源：https://mp.weixin.qq.com/s/zB7dM9xt26c6Z04PvYfOmQ