真香，聊聊 RocketMQ 5.0 的 POP 消費(fèi)模式！
大家好，我是君哥。
大家都知道，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ù)肀г圃?，客戶端要改造成無狀態(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)榭梢?，再次被消費(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);
}
}
}3 總結(jié)
POP 客戶端有很多的優(yōu)勢，總結(jié)如下：
- 無狀態(tài)，更好地?fù)肀г圃?/li>
- 計(jì)算相關(guān)的功能下移到 Proxy，更加輕量級；
- 消費(fèi)能力擴(kuò)展不受 MessageQueue 數(shù)量的限制；
- 消費(fèi)者 hang 住，并不會(huì)導(dǎo)致消息積壓。
