日本综合一区二区|亚洲中文天堂综合|日韩欧美自拍一区|男女精品天堂一区|欧美自拍第6页亚洲成人精品一区|亚洲黄色天堂一区二区成人|超碰91偷拍第一页|日韩av夜夜嗨中文字幕|久久蜜综合视频官网|精美人妻一区二区三区

RELATEED CONSULTING
相關(guān)咨詢
選擇下列產(chǎn)品馬上在線溝通
服務(wù)時間：8:30-17:00
你可能遇到了下面的問題
關(guān)閉右側(cè)工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
五張圖帶你理解RocketMQ順序消息實現(xiàn)機制

大家好，我是君哥。今天聊一聊 RocketMQ 的順序消息實現(xiàn)機制。

為廊坊等地區(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ù)。我們深信只要達到每一位用戶的要求，就會得到認可，從而選擇與我們長期合作。這樣，我們也可以走得更遠！

在有些場景下，使用 MQ 需要保證消息的順序性，比如在電商系統(tǒng)中，用戶提交訂單、支付訂單、訂單出庫這 3 個消息應(yīng)該保證順序性，如下圖：

對于 RocketMQ 來說，主要是通過 Producer 和 Consumer 來保證消息順序的。

1、Producer

下面代碼是 Producer 發(fā)送順序消息的官方示例：

public static void main(String[] args) throws UnsupportedEncodingException {
 try {
  DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
  producer.start();
  String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
  for (int i = 0; i < 100; i++) {
   int orderId = i % 10;
   Message msg =
    new Message("TopicTestjjj", tags[i % tags.length], "KEY" + i,
     ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
   SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List mqs, Message msg, Object arg) {
     Integer id = (Integer) arg;
     int index = id % mqs.size();
     return mqs.get(index);
    }
   }, orderId);

   System.out.printf("%s%n", sendResult);
  }
  producer.shutdown();
 } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
  e.printStackTrace();
 }
}

跟發(fā)送并發(fā)消息不一樣的是，發(fā)送消息時傳入了 MessageQueueSelector，這里可以指定消息發(fā)送到固定的 MessageQueue。

注意：上面的代碼把 orderId 相同的消息都會發(fā)送到同一個 MessageQueue，這樣同一個 orderId 的消息是有序的，這也叫做局部有序。對應(yīng)的另一種是全局有序，這需要把所有的消息都發(fā)到同一個 MessageQueue。

下面再來看一下發(fā)送的代碼：

private SendResult sendSelectImpl(
 Message msg,
 MessageQueueSelector selector,
 Object arg,
 final CommunicationMode communicationMode,
 final SendCallback sendCallback, final long timeout
) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
 //省略部分邏輯
 TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
 if (topicPublishInfo != null && topicPublishInfo.ok()) {
  MessageQueue mq = null;
  try {
   List messageQueueList =
    mQClientFactory.getMQAdminImpl().parsePublishMessageQueues(topicPublishInfo.getMessageQueueList());
   Message userMessage = MessageAccessor.cloneMessage(msg);
   String userTopic = NamespaceUtil.withoutNamespace(userMessage.getTopic(), mQClientFactory.getClientConfig().getNamespace());
   userMessage.setTopic(userTopic);

   mq = mQClientFactory.getClientConfig().queueWithNamespace(selector.select(messageQueueList, userMessage, arg));
  } catch (Throwable e) {
   throw new MQClientException("select message queue threw exception.", e);
  }
  //省略部分邏輯
  if (mq != null) {
   return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, null, timeout - costTime);
  } else {
   throw new MQClientException("select message queue return null.", null);
  }
 }
    //省略部分邏輯
}

可以看到，在發(fā)送的時候，使用 MessageQueueSelector 選擇一個 MessageQueue，然后發(fā)送消息到這個 MessageQueue。對于并發(fā)消息，這里不傳 MessageQueueSelector，如果發(fā)送方法沒有指定 MessageQueue，就會按照默認的策略選擇一個。

2、Consumer

以 RocketMQ 推模式為例，消費者會注冊一個監(jiān)聽器，進行消息的拉取和消費處理，下面的 UML 類圖顯示了調(diào)用關(guān)系：

上圖中包含了對順序消息和對并發(fā)消息的處理。其中 MessageListenerOrderly 和 ConsumeMessageOrderlyService 對順序消息進行處理。跟并發(fā)消息不一樣的是，順序消息定義了一個 MessageQueueLock 類，這個類保存了每個 MessageQueue 對應(yīng)的鎖，代碼如下：

private ConcurrentMap mqLockTable = new ConcurrentHashMap();

下面代碼是順序消費的官方示例：

public static void main(String[] args) throws MQClientException {
 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
 consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
 consumer.subscribe("TopicTest", "TagA || TagC || TagD");
 consumer.registerMessageListener(new MessageListenerOrderly() {
  AtomicLong consumeTimes = new AtomicLong(0);
  @Override
  public ConsumeOrderlyStatus consumeMessage(List msgs, ConsumeOrderlyContext context) {
   context.setAutoCommit(true);
   System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
   this.consumeTimes.incrementAndGet();
   if ((this.consumeTimes.get() % 2) == 0) {
    return ConsumeOrderlyStatus.SUCCESS;
   } else if ((this.consumeTimes.get() % 5) == 0) {
    context.setSuspendCurrentQueueTimeMillis(3000);
    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
   }
   return ConsumeOrderlyStatus.SUCCESS;
  }
 });
 consumer.start();
 System.out.printf("Consumer Started.%n");
}

下面看一下順序消息的消費端處理邏輯。

（1）注冊監(jiān)聽

上面的代碼定義了順序消息監(jiān)聽器 MessageListenerOrderly，并且注冊到 DefaultMQPushConsumer，這個注冊同時也注冊到了 DefaultMQPushConsumerImpl。

（2）PushConsumer 初始化

在 DefaultMQPushConsumerImpl 類初始化的時候，會判斷注冊的 MessageListener 是不是 MessageListenerOrderly，如果是，就把 consumeOrderly 變量設(shè)置為 true，以此來標記是順序消息拉取還是并發(fā)消息拉取。然后把 ConsumeMessageService 初始化為 ConsumeMessageOrderlyService。

（3）鎖定 mq

要保證消息的順序性，就需要保證同一個 MessageQueue 只能被同一個 Consumer 消費。

ConsumeMessageOrderlyService 初始化的時候，會啟動一個定時任務(wù)，周期性（默認 20s）地向 Broker 發(fā)送鎖定消息（請求類型 LOCK_BATCH_MQ），Broker 收到后，就會把 MessageQueue、group 和 clientId 進行綁定，這樣其他客戶端就不能從這個 MessageQueue 拉取消息。

注意：Broker 的鎖定是有過期時間的，默認 60s，可以配置，鎖定過期后，有可能被其他 Consumer 進行消費。

Broker 端鎖結(jié)構(gòu)如下圖：

（4）拉取消息

消費者啟動時，啟動消費拉取線程 PullMessageService，里面死循環(huán)不停地從 Broker 拉取消息。這里調(diào)用了 DefaultMQPushConsumerImpl 類的 pullMessage 方法。這里拉取消息的邏輯跟并發(fā)消息邏輯是一樣的。

拉取到消息后，調(diào)用 PullCallback 的 onSuccess 方法處理結(jié)果，這里調(diào)用了 ConsumeMessageOrderlyService 的 submitConsumeRequest 方法，里面用線程池提交了 ConsumeRequest 線程。

PullCallback pullCallback = new PullCallback() {
 @Override
 public void onSuccess(PullResult pullResult) {
  if (pullResult != null) {
   pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,
    subscriptionData);
   switch (pullResult.getPullStatus()) {
    case FOUND:
     //省略
     if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {
      DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
     } else {
      //省略
      boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
      DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
       pullResult.getMsgFoundList(),
       processQueue,
       pullRequest.getMessageQueue(),
       dispatchToConsume);
                        //省略
     }
     //省略
     break;
    //省略
   }
  }
 }
    //省略
};

上面拉取到消息后，先把消息放到了 ProcessQueue，然后調(diào)用了 submitConsumeRequest 方法。跟并發(fā)消息處理方式不同的是，submitConsumeRequest 方法并沒有處理拉取到的消息，而真正處理的時候是從 ProcessQueue 獲取。

（5）處理消息

處理消息的邏輯在 ConsumeMessageOrderlyService 的內(nèi)部類 ConsumeRequest，這是一個線程類，run 方法如下：

public void run() {
 //省略部分邏輯
 //1.獲取到 MessageQueueLock 對應(yīng)的鎖
 final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
 synchronized (objLock) {
  if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
   || (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
   final long beginTime = System.currentTimeMillis();
   for (boolean continueConsume = true; continueConsume; ) {
    //省略延后執(zhí)行的邏輯
    final int consumeBatchSize =
     ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
                //2.從 processQueue 拉取消息
    List msgs = this.processQueue.takeMessages(consumeBatchSize);
    if (!msgs.isEmpty()) {
     final ConsumeOrderlyContext context = new ConsumeOrderlyContext(this.messageQueue);

     ConsumeOrderlyStatus status = null;
                    //省略部分邏輯
     boolean hasException = false;
     try {
         //3.獲取處理鎖
      this.processQueue.getConsumeLock().lock();
      //4.執(zhí)行消費處理邏輯
      status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
     } catch (Throwable e) {
      log.warn(String.format("consumeMessage exception: %s Group: %s Msgs: %s MQ: %s",
       RemotingHelper.exceptionSimpleDesc(e),
       ConsumeMessageOrderlyService.this.consumerGroup,
       msgs,
       messageQueue), e);
      hasException = true;
     } finally {
         //5.釋放處理鎖
      this.processQueue.getConsumeLock().unlock();
     }
     //省略部分邏輯
     continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult(msgs, status, context, this);
    } else {
     continueConsume = false;
    }
   }
  } else {
   //省略部分邏輯
   ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100);
  }
 }
}

上面的代碼總結(jié)一下，Consumer 消費消息的邏輯如下：

  1. 對 MessageQueueLock 進行加鎖，這樣就保證只有一個線程在處理當(dāng)前 MessageQueue。
  2. 從 ProcessQueue 拉取一批消息。
  3. 獲取 ProcessQueue 鎖，這樣保證了只有當(dāng)前線程可以進行消息處理，同時也可以防止 Rebalance 線程把當(dāng)前處理的 MessageQueue 移除掉。
  4. 執(zhí)行消費處理邏輯。
  5. 釋放 ProcessQueue 處理鎖；6.processConsumeResult 方法更新消息偏移量。

注意：ProcessQueue 中的鎖是 ReentrantLock。

3、重試

跟并發(fā)消息不一樣的是，順序消息消費失敗后并不會把消息發(fā)送到 Broker，而是直接在 Consumer 端進行重試，如果重試次數(shù)超過了最大重試次數(shù)（16 次），則發(fā)送到 Broker，Broker 則將消息推入死信隊列。如下圖：

4、總結(jié)

RocketMQ 順序消息的原理是在 Producer 端把一批需要保證順序的消息發(fā)送到同一個 MessageQueue，Consumer 端則通過加鎖的機制來保證消息消費的順序性，Broker 端通過對 MessageQueue 進行加鎖，保證同一個 MessageQueue 只能被同一個 Consumer 進行消費。

根據(jù)實現(xiàn)原理可以看到，RocketMQ 的順序消息可能存在兩個坑：

  1. 有順序性的消息需要發(fā)送到同一個 MessageQueue，可能導(dǎo)致單個 MessageQueue 消息量很大，而 Consumer 端消費的時候只能單線程消費，很可能導(dǎo)致當(dāng)前 MessageQueue 消息積壓。
  2. 如果順序消息 MessageQueue 所在的 broker 掛了，這時 Producer 只能把消息發(fā)送到其他 Broker 的 MessageQueue 上，而如果新的 MessageQueue 被其他 Consumer 消費，這樣兩個 Consumer 消費的消息就不能保證順序性了。如下圖：

Broker1 發(fā)生故障，把訂單出庫的消息發(fā)送到了 Broker2，由 Consumer2 來進行消費，消息順序很可能會錯亂。


分享名稱：五張圖帶你理解RocketMQ順序消息實現(xiàn)機制
網(wǎng)頁URL：http://www.dlmjj.cn/article/dhgcggc.html