RocketMQ 5.0 的 POP 消费模式，很香！

2023-06-0809:10:00云计算与物联网Comments1,217 views字数 5947阅读模式

RocketMQ 消费模式有 PULL 模式和 PUSH 模式，不过本质上都是 PULL 模式，而在实际使用时，一般使用 PUSH 模式。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

不过，RocketMQ 的 PUSH 模式有明显的不足，主要体现在以下几个方面：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

  1. 消息积压了，增加消费者不一定能解决。PUSH 模式如下图：
RocketMQ 5.0 的 POP 消费模式，很香！

上面的图中，消费组中的消费者每个消费者消费两个 MessageQueue，这种情况下，增加消费者是可以提高消费能力的。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

但是下面这张图，每个消费者消费一个 MessageQueue，因为同一个 MessageQueue 只能被同一个消费组中的一个消费者消费，所以增加消费者并不能提高消费能力。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

RocketMQ 5.0 的 POP 消费模式，很香！文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

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

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

RocketMQ 5.0 为了解决 PUSH Consumer 上面的问题，引入了 POP Consumer。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

1 POP 客户端

POP 模式的客户端引入的背景是 RocketMQ 5.0 为了更好地拥抱云原生，客户端要改造成无状态的轻量级客户端，RocketMQ 4.x 中客户端具有的负载均衡、权限管理、消费管理等功能都从客户端移动到了 Proxy。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

POP 消费模式如下图：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

RocketMQ 5.0 的 POP 消费模式，很香！

四个消费者都可以消费 Broker1 和 Broker2 上面的所有队列，这样即使某一个消费者 hang 住了，其他消费者也可以消费，并不会造成消息积压。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

同时，从上图中可以看到，POP 客户端还有一个优势，增加消费者数量是可以提高消费能力的，不受 MessageQueue 数量和消费者数量的限制。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

跟 PUSH 模式相比，POP 模式拉取到消息后，会设置一个 POP_CK 属性，代码如下：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

//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 等参数，通过这个属性可以唯一标识一条消息了。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

从上面的代码还可以看到，responseHeader 中有一个 invisibleTime 属性，这个属性的作用是消费者通过 POP 模式拉取到一条消息后，这段时间（invisibleTime）内这条消息在 Broker 端是不可见的，消费者再次拉取就不会重复拉取到。但是如果过了这段时间，消费者还没有给 Broker 返回 ACK，这条消息会变为可见，再次被消费者拉取到。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

消费完成后，向 Broker 发送 ACK 消息，见下面代码：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

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 上面拉取，有没有可能会重复消费呢？文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

Broker 收到消息拉取请求，从 MessageStore 拉取消息时，首先会给 MessageQueue 进行加锁，加锁成功后，才会拉取消息，这是其他客户端来拉取时就会加锁失败。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

//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 放入缓存，代码如下：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

//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 从缓存中移除。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

如果 Broker 一直没有收到 ACK，则会把 CheckPoint 从缓存中移除，同时把 CheckPoint 发送给 MessageStore，由 MessageStore 发送到重试队列。代码如下：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

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

3 总结

POP 客户端有很多的优势，总结如下：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

  1. 无状态，更好地拥抱云原生；
  2. 计算相关的功能下移到 Proxy，更加轻量级；
  3. 消费能力扩展不受 MessageQueue 数量的限制；
  4. 消费者 hang 住，并不会导致消息积压。

来源于君哥聊技术 ，作者朱晋君文章源自菜鸟学院-https://www.cainiaoxueyuan.com/yunda/45826.html

  • 本站内容整理自互联网，仅提供信息存储空间服务，以方便学习之用。如对文章、图片、字体等版权有疑问，请在下方留言，管理员看到后，将第一时间进行处理。
  • 转载请务必保留本文链接：https://www.cainiaoxueyuan.com/yunda/45826.html

Comment

匿名网友 填写信息

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定