日韩无码专区无码一级三级片|91人人爱网站中日韩无码电影|厨房大战丰满熟妇|AV高清无码在线免费观看|另类AV日韩少妇熟女|中文日本大黄一级黄色片|色情在线视频免费|亚洲成人特黄a片|黄片wwwav色图欧美|欧亚乱色一区二区三区

大家好，我是君哥。

為蓬萊等地區(qū)用戶提供了全套網(wǎng)頁設(shè)計制作服務(wù)，及蓬萊網(wǎng)站建設(shè)行業(yè)解決方案。主營業(yè)務(wù)為網(wǎng)站建設(shè)、成都做網(wǎng)站、蓬萊網(wǎng)站設(shè)計，以傳統(tǒng)方式定制建設(shè)網(wǎng)站，并提供域名空間備案等一條龍服務(wù)，秉承以專業(yè)、用心的態(tài)度為用戶提供真誠的服務(wù)。我們深信只要達到每一位用戶的要求，就會得到認可，從而選擇與我們長期合作。這樣，我們也可以走得更遠！

今天來分享 RocketMQ 中一個非常重要又不太好理解的知識點-ProcessQueue。

一句話概括，ProcessQueue 就是 MessageQueue 的消費快照?？聪旅孢@張圖：

1 ProcessQueue 構(gòu)建

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，數(shù)據(jù)結(jié)構(gòu)如下：

可以看到，在 processQueueTable 這個數(shù)據(jù)結(jié)構(gòu)上維護了 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;
    }
   }
  }
 }
    //創(chuàng)建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后再創(chuàng)建一個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 拉取消息

上一節(jié)中構(gòu)建 ProcessQueue 后，會再創(chuàng)建一個 PullRequest，這個 PullRequest 封裝了 MessageQueue 和 ProcessQueue，創(chuàng)建成功后被放到了 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 拉取到消息后，會進行如下處理：

1. 對拉取到的消息根據(jù) TAG 再次
   進行過濾；
2. 更新 PullRequest 下次拉取的偏移量 nextOffset；
3. 把拉取的消息封裝到 ProcessQueue 的 msgTreeMap（
   ?
   放到 msgTreeMap 之前首先要獲取到寫鎖 treeMapLock
   ?
   ）；
4. 封裝 ConsumeRequest 進行消息消費;
5. 封裝消息拉取請求再次進行拉取。

代碼如下：

//DefaultMQPushConsumerImpl.java
public void onSuccess(PullResult pullResult) {
 if (pullResult != null) {
     //1. 對拉取到的消息根據(jù) 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 消費消息

在上一節(jié)提到過，拉取到消息后，會把消息封裝成一個 ConsumeRequest,這個線程類會調(diào)用消費者定義的 MessageListener 進行消費處理?？匆幌略创a：

//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);
 }
}

消息消費成功后，會調(diào)用 processConsumeResult 方法進行結(jié)果處理。對于廣播模式，發(fā)送失敗后不會做重試，相當于把消息丟棄，而對于集群模式，消費失敗的消息會發(fā)送到 Broker 端等待消費者重新拉取進行重試。

消費結(jié)果處理完后，消費成功的消息會從 ProcessQueue 的 msgTreeMap 中移除（需要獲取到寫鎖 treeMapLock），同時從 msgTreeMap 中獲取最小的 Offset 來更新對應(yīng) 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);
    //消費失敗的，發(fā)送回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 緩存消息數(shù)量

如果消費者緩存的消息數(shù)量大于 RocketMQ 配置的閾值（默認 1000），就會觸發(fā)延遲拉取，而消費者緩存的消息數(shù)量就來自 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），就會觸發(fā)延遲拉取，而消費者緩存的消息大小就來自 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），就會觸發(fā)延遲拉取，而消息間隔就來自 ProcessQueue，看下面代碼：

if (!this.consumeOrderly) {
 if (processQueue.getMaxSpan() > this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan()) {
  this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
  return;
 }
}

4.4 獲取鎖失敗

對于順序消息，如果獲取鎖失敗，也會觸發(fā)延遲拉取，而判斷獲取鎖是否成功，也是在 ProcessQueue，看下面代碼：

if (processQueue.isLocked()) {
 //...
} else {
 this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
}

5 總結(jié)

ProcessQueue 是 MessageQueue 的消費快照，可以協(xié)助消費者進行消息拉取、消息消費、更新偏移量、限流。最后，看一下 ProcessQueue 的數(shù)據(jù)結(jié)構(gòu)：

網(wǎng)頁標題：老弟問我，RocketMQ中的ProcessQueue怎么理解？
分享路徑：http://www.5511xx.com/article/dppdopi.html