真香，聊聊 RocketMQ 5.0 的 POP 消费模式！

大家好，我是君哥。

大家都知道，RocketMQ 消费模式有 PULL 模式和 PUSH 模式，不过本质上都是 PULL 模式，而在实际使用时，一般使用 PUSH 模式。

不过，RocketMQ 的 PUSH 模式有明显的不足，主要体现在以下几个方面：

消息积压了，增加消费者不一定能解决。PUSH 模式如下图：

上面的图中，消费组中的消费者每个消费者消费两个 MessageQueue，这种情况下，增加消费者是可以提高消费能力的。

但是下面这张图，每个消费者消费一个 MessageQueue，因为同一个 MessageQueue 只能被同一个消费组中的一个消费者消费，所以增加消费者并不能提高消费能力。

客户端的处理逻辑比较多，比如负载均衡、offset 管理、消费失败后的处理（比如失败消息发送回 Broker），这些逻辑都在客户端。
如果再支持其他语言，客户端会变得越来越重。
消费者机器 hang 住，可能会导致消息积压，如下图：

通过客户端负责均衡，MessageQueue0 这个队列分配给了 Consumer0 进行独占消费，如果 Consumer0 这个消费者 hang 住了，但是服务没有挂，不能从 Name Server 中下线，因为 Consumer0 拉取到的消息不能消费，也就不能给 Broker 发送更新 Offset 的请求，最终导致消息积压。这种情况只能手动让 Consumer0 下线或者让 Consumer0 重启。

RocketMQ 5.0 为了解决 PUSH Consumer 上面的问题，引入了 POP Consumer。

1 POP 客户端

POP 模式的客户端引入的背景是 RocketMQ 5.0 为了更好地拥抱云原生，客户端要改造成无状态的轻量级客户端，RocketMQ 4.x 中客户端具有的负载均衡、权限管理、消费管理等功能都从客户端移动到了 Proxy。

POP 消费模式如下图：

四个消费者都可以消费 Broker1 和 Broker2 上面的所有队列，这样即使某一个消费者 hang 住了，其他消费者也可以消费，并不会造成消息积压。

同时，从上图中可以看到，POP 客户端还有一个优势，增加消费者数量是可以提高消费能力的，不受 MessageQueue 数量和消费者数量的限制。

跟 PUSH 模式相比，POP 模式拉取到消息后，会设置一个 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 等参数，通过这个属性可以唯一标识一条消息了。

从上面的代码还可以看到，responseHeader 中有一个 invisibleTime 属性，这个属性的作用是消费者通过 POP 模式拉取到一条消息后，这段时间（invisibleTime）内这条消息在 Broker 端是不可见的，消费者再次拉取就不会重复拉取到。但是如果过了这段时间，消费者还没有给 Broker 返回 ACK，这条消息会变为可见，再次被消费者拉取到。

消费完成后，向 Broker 发送 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

从上面的介绍可以看到，每个消费者都可以从 Broker 的所有 MessageQueue 上拉取消息，那如果多个消费者都从一个 MessageQueue 上面拉取，有没有可能会重复消费呢？

Broker 收到消息拉取请求，从 MessageStore 拉取消息时，首先会给 MessageQueue 进行加锁，加锁成功后，才会拉取消息，这是其他客户端来拉取时就会加锁失败。

//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 拉取到消息后，会定义一个 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 收到消费者发来的 ACK 后，会把 CheckPoint 从缓存中移除。

如果 Broker 一直没有收到 ACK，则会把 CheckPoint 从缓存中移除，同时把 CheckPoint 发送给 MessageStore，由 MessageStore 发送到重试队列。代码如下：

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