久久免费大香蕉,91久久精品无码一区二区三区,国产无码A V

大家都知道，RocketMQ 消費(fèi)模式有 PULL 模式和 PUSH 模式，不過本質(zhì)上都是 PULL 模式，而在實(shí)際使用時(shí)，一般使用 PUSH 模式。

不過，RocketMQ 的 PUSH 模式有明顯的不足，主要體現(xiàn)在以下幾個(gè)方面：

消息積壓了，增加消費(fèi)者不一定能解決。PUSH 模式如下圖：

上面的圖中，消費(fèi)組中的消費(fèi)者每個(gè)消費(fèi)者消費(fèi)兩個(gè) MessageQueue，這種情況下，增加消費(fèi)者是可以提高消費(fèi)能力的。

但是下面這張圖，每個(gè)消費(fèi)者消費(fèi)一個(gè) MessageQueue，因?yàn)橥粋€(gè) MessageQueue 只能被同一個(gè)消費(fèi)組中的一個(gè)消費(fèi)者消費(fèi)，所以增加消費(fèi)者并不能提高消費(fèi)能力。

客戶端的處理邏輯比較多，比如負(fù)載均衡、offset 管理、消費(fèi)失敗后的處理（比如失敗消息發(fā)送回 Broker），這些邏輯都在客戶端。
如果再支持其他語言，客戶端會(huì)變得越來越重。
消費(fèi)者機(jī)器 hang 住，可能會(huì)導(dǎo)致消息積壓，如下圖：

通過客戶端負(fù)責(zé)均衡，MessageQueue0 這個(gè)隊(duì)列分配給了 Consumer0 進(jìn)行獨(dú)占消費(fèi)，如果 Consumer0 這個(gè)消費(fèi)者 hang 住了，但是服務(wù)沒有掛，不能從 Name Server 中下線，因?yàn)?Consumer0 拉取到的消息不能消費(fèi)，也就不能給 Broker 發(fā)送更新 Offset 的請求，最終導(dǎo)致消息積壓。這種情況只能手動(dòng)讓 Consumer0 下線或者讓 Consumer0 重啟。

RocketMQ 5.0 為了解決 PUSH Consumer 上面的問題，引入了 POP Consumer。

1 POP 客戶端

POP 模式的客戶端引入的背景是 RocketMQ 5.0 為了更好地?fù)肀г圃蛻舳艘脑斐蔁o狀態(tài)的輕量級客戶端，RocketMQ 4.x 中客戶端具有的負(fù)載均衡、權(quán)限管理、消費(fèi)管理等功能都從客戶端移動(dòng)到了 Proxy。

POP 消費(fèi)模式如下圖：

四個(gè)消費(fèi)者都可以消費(fèi) Broker1 和 Broker2 上面的所有隊(duì)列，這樣即使某一個(gè)消費(fèi)者 hang 住了，其他消費(fèi)者也可以消費(fèi)，并不會(huì)造成消息積壓。

同時(shí)，從上圖中可以看到，POP 客戶端還有一個(gè)優(yōu)勢，增加消費(fèi)者數(shù)量是可以提高消費(fèi)能力的，不受 MessageQueue 數(shù)量和消費(fèi)者數(shù)量的限制。

跟 PUSH 模式相比，POP 模式拉取到消息后，會(huì)設(shè)置一個(gè) POP_CK 屬性，代碼如下：

//MQClientAPIImpl.JAVA
if (requestHeader instanceof PopMessageRequestHeader) {
 if (startOffsetInfo == null) {
  // we should set the check point info to extraInfo field , if the command is popMsg
  // find pop ck offset
  String key = messageExt.getTopic() + messageExt.getQueueId();
  if (!map.containsKey(messageExt.getTopic() + messageExt.getQueueId())) {
   map.put(key, ExtraInfoUtil.buildExtraInfo(messageExt.getQueueOffset(), responseHeader.getPopTime(), responseHeader.getInvisibleTime(), responseHeader.getReviveQid(),
    messageExt.getTopic(), brokerName, messageExt.getQueueId()));

  }
  messageExt.getProperties().put(MessageConst.PROPERTY_POP_CK, map.get(key) + MessageConst.KEY_SEPARATOR + messageExt.getQueueOffset());
 } else {
  String queueIdKey = ExtraInfoUtil.getStartOffsetInfoMapKey(messageExt.getTopic(), messageExt.getQueueId());
  String queueOffsetKey = ExtraInfoUtil.getQueueOffsetMapKey(messageExt.getTopic(), messageExt.getQueueId(), messageExt.getQueueOffset());
  int index = sortMap.get(queueIdKey).indexOf(messageExt.getQueueOffset());
  Long msgQueueOffset = msgOffsetInfo.get(queueIdKey).get(index);

  messageExt.getProperties().put(MessageConst.PROPERTY_POP_CK,
   ExtraInfoUtil.buildExtraInfo(startOffsetInfo.get(queueIdKey), responseHeader.getPopTime(), responseHeader.getInvisibleTime(),
    responseHeader.getReviveQid(), messageExt.getTopic(), brokerName, messageExt.getQueueId(), msgQueueOffset)
  );
  //...
 }
}

可以看到，POP_CK 屬性包含了 brokerName、Topic、QueueId、offset 等參數(shù)，通過這個(gè)屬性可以唯一標(biāo)識一條消息了。

從上面的代碼還可以看到，responseHeader 中有一個(gè) invisibleTime 屬性，這個(gè)屬性的作用是消費(fèi)者通過 POP 模式拉取到一條消息后，這段時(shí)間（invisibleTime）內(nèi)這條消息在 Broker 端是不可見的，消費(fèi)者再次拉取就不會(huì)重復(fù)拉取到。但是如果過了這段時(shí)間，消費(fèi)者還沒有給 Broker 返回 ACK，這條消息會(huì)變?yōu)榭梢姡俅伪幌M(fèi)者拉取到。

消費(fèi)完成后，向 Broker 發(fā)送 ACK 消息，見下面代碼：

public void ackMessageAsync(
 final String addr,
 final long timeOut,
 final AckCallback ackCallback,
 final AckMessageRequestHeader requestHeader //
) throws RemotingException, MQBrokerException, InterruptedException {
 final RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.ACK_MESSAGE, requestHeader);
 this.remotingClient.invokeAsync(addr, request, timeOut, new BaseInvokeCallback(MQClientAPIImpl.this) {

  @Override
  public void onComplete(ResponseFuture responseFuture) {
   RemotingCommand response = responseFuture.getResponseCommand();
   if (response != null) {
    try {
     AckResult ackResult = new AckResult();
     if (ResponseCode.SUCCESS == response.getCode()) {
      ackResult.setStatus(AckStatus.OK);
     } //...
     assert ackResult != null;
     ackCallback.onSuccess(ackResult);
    } //...
   } else {
    //...
   }

  }
 });
}

2. Broker

從上面的介紹可以看到，每個(gè)消費(fèi)者都可以從 Broker 的所有 MessageQueue 上拉取消息，那如果多個(gè)消費(fèi)者都從一個(gè) MessageQueue 上面拉取，有沒有可能會(huì)重復(fù)消費(fèi)呢？

Broker 收到消息拉取請求，從 MessageStore 拉取消息時(shí)，首先會(huì)給 MessageQueue 進(jìn)行加鎖，加鎖成功后，才會(huì)拉取消息，這是其他客戶端來拉取時(shí)就會(huì)加鎖失敗。

//PopMessageProcessor.java
String lockKey = topic + PopAckConstants.SPLIT + requestHeader.getConsumerGroup() + PopAckConstants.SPLIT + queueId;
long offset = getPopOffset(topic, requestHeader, queueId, false, lockKey);
if (!queueLockManager.tryLock(lockKey)) {
 restNum = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId) - offset + restNum;
 return restNum;
}

Broker 從 MessageStore 拉取到消息后，會(huì)定義一個(gè) CheckPoint 放入緩存，代碼如下：

//PopMessageProcessor.java
private long popMsgFromQueue(boolean isRetry, GetMessageResult getMessageResult,
 PopMessageRequestHeader requestHeader, int queueId, long restNum, int reviveQid,
 Channel channel, long popTime,
 ExpressionMessageFilter messageFilter, StringBuilder startOffsetInfo,
 StringBuilder msgOffsetInfo, StringBuilder orderCountInfo) {
 String topic = isRetry ? KeyBuilder.buildPopRetryTopic(requestHeader.getTopic(),
  requestHeader.getConsumerGroup()) : requestHeader.getTopic();
 String lockKey =
  topic + PopAckConstants.SPLIT + requestHeader.getConsumerGroup() + PopAckConstants.SPLIT + queueId;
 //...
 offset = getPopOffset(topic, requestHeader, queueId, true, lockKey);
 GetMessageResult getMessageTmpResult = null;
 try {
  //...

  restNum = getMessageTmpResult.getMaxOffset() - getMessageTmpResult.getNextBeginOffset() + restNum;
  if (!getMessageTmpResult.getMessageMapedList().isEmpty()) {

   if (isOrder) {
    //...
   } else {
    AppendCheckPoint(requestHeader, topic, reviveQid, queueId, offset, getMessageTmpResult, popTime, this.brokerController.getBrokerConfig().getBrokerName());
   }
  } //...
 } //...
 return restNum;
}

Broker 收到消費(fèi)者發(fā)來的 ACK 后，會(huì)把 CheckPoint 從緩存中移除。

如果 Broker 一直沒有收到 ACK，則會(huì)把 CheckPoint 從緩存中移除，同時(shí)把 CheckPoint 發(fā)送給 MessageStore，由 MessageStore 發(fā)送到重試隊(duì)列。代碼如下：

boolean removeCk = !this.serving;
 // ck will be timeout
 if (point.getReviveTime() - now < brokerController.getBrokerConfig().getPopCkStayBufferTimeOut()) {
  removeCk = true;
 }

 // the time stayed is too long
 if (now - point.getPopTime() > brokerController.getBrokerConfig().getPopCkStayBufferTime()) {
  removeCk = true;
 }

 // double check
 if (removeCk) {
  // put buffer ak to store
  if (pointWrapper.getReviveQueueOffset() < 0) {
   putCkToStore(pointWrapper, false);
  }
 }
}