大家好,我是君哥。
大家都知道,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);
}
}
}
3 总结
POP 客户端有很多的优势,总结如下:
-
无状态,更好地拥抱云原生;
-
计算相关的功能下移到 Proxy,更加轻量级;
-
消费能力扩展不受 MessageQueue 数量的限制;
-
消费者 hang 住,并不会导致消息积压。