一文帶你理解 RocketMQ 廣播模式實現(xiàn)機制
大家好，我是君哥。今天聊聊 RocketMQ 的廣播消息實現(xiàn)機制。
RocketMQ 有兩種消費模式，集群模式和廣播模式。
集群模式是指 RocketMQ 中的一條消息只能被同一個消費者組中的一個消費者消費。如下圖，Producer 向 TopicTest 這個 Topic 并發(fā)寫入 3 條新消息，分別被分配到了 MessageQueue1~MessageQueue3 這 3 個隊列，然后 Group 中的三個 Consumer 分別消費了一條消息：
廣播模式是 RocketMQ 中的消息會被消費組中的每個消費者都消費一次，如下圖：
使用 RocketMQ 的廣播模式時，需要在消費端進行定義，下面是一段官方示例：
public static void main(String[] args) throws InterruptedException, MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_1");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.setMessageModel(MessageModel.BROADCASTING);
consumer.subscribe("TopicTest", "TagA || TagC || TagD");
consumer.registerMessageListener(new MessageListenerConcurrently() {
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.printf("Broadcast Consumer Started.%n");
}
從代碼中可以看到，在定義 Consumer 時，通過 messageModel 這個屬性指定消費模式，這里指定為 BROADCASTING，也就啟動了廣播模式的消費者。
1、消費者啟動
以 RocketMQ 推模式為例，看一下消費者調(diào)用關(guān)系類圖：
DefaultMQPushConsumer 作為啟動入口類，它的 start 方法調(diào)用了 DefaultMQPushConsumerImpl 類的 start 方法，下面重點看一下這個方法。
（1）拷貝訂閱關(guān)系
start 方法中調(diào)用了 copySubscription 方法，代碼如下：
private void copySubscription() throws MQClientException {
try {
//拷貝訂閱關(guān)系
switch (this.defaultMQPushConsumer.getMessageModel()) {
case BROADCASTING:
break;
case CLUSTERING:
final String retryTopic = MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup());
SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(retryTopic, SubscriptionData.SUB_ALL);
this.rebalanceImpl.getSubscriptionInner().put(retryTopic, subscriptionData);
break;
default:
break;
}
} catch (Exception e) {
throw new MQClientException("subscription exception", e);
}
}
這里的代碼有一點需要注意：集群模式會創(chuàng)建一個重試 Topic 的訂閱關(guān)系，而廣播模式是不會創(chuàng)建這個訂閱關(guān)系的。也就是說廣播模式不考慮重試。
（2）初始化偏移量
下面是初始化 offset 的代碼：
if (this.defaultMQPushConsumer.getOffsetStore() != null) {
this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
} else {
switch (this.defaultMQPushConsumer.getMessageModel()) {
case BROADCASTING:
this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
break;
case CLUSTERING:
this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
break;
default:
break;
}
this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
}
從上面的代碼可以看到，廣播模式使用了 LocalFileOffsetStore，也就是說偏移量保存在客戶端本地，除了在內(nèi)存中會保存，在本地文件中也會保存。
2、消息拉取
ConsumeMessageService 是真正拉取消息的地方，消費者初始化時會初始化 ConsumeMessageService，并且這里會區(qū)分并發(fā)消息還是順序消息。
（1）順序消息
在集群模式下，需要獲取到 processQueue 的鎖才會拉取消息，而在廣播模式下，不用獲取鎖，直接就可以拉取消息。判斷邏輯如下：
//ConsumeMessageOrderlyService.ConsumeRequest
final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
synchronized (objLock) {
if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
|| (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
}
}
這里有個疑問，對于順序消息，獲取鎖是必須的，這樣才能保證一個 processQueue 只能由一個線程進行處理，從而保證消費的順序性。那對于廣播模式，為什么不用獲取 processQueue 的鎖呢？難道廣播模式不支持順序消息？
（2）并發(fā)消息
對于并發(fā)消息，廣播模式不同的是，對消費結(jié)果的處理。集群模式消費失敗后需要把消息發(fā)送回 Broker 等待再次被拉取，而廣播模式則不需要重試。代碼如下：
//ConsumeMessageConcurrentlyService.rocessConsumeResult
switch (this.defaultMQPushConsumer.getMessageModel()) {
case BROADCASTING:
for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {
MessageExt msg = consumeRequest.getMsgs().get(i);
log.warn("BROADCASTING, the message consume failed, drop it, {}", msg.toString());
}
break;
case CLUSTERING:
List<MessageExt> msgBackFailed = new ArrayList<MessageExt>(consumeRequest.getMsgs().size());
for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {
MessageExt msg = consumeRequest.getMsgs().get(i);
boolean result = this.sendMessageBack(msg, context);
if (!result) {
msg.setReconsumeTimes(msg.getReconsumeTimes() + 1);
msgBackFailed.add(msg);
}
}
if (!msgBackFailed.isEmpty()) {
consumeRequest.getMsgs().removeAll(msgBackFailed);
this.submitConsumeRequestLater(msgBackFailed, consumeRequest.getProcessQueue(), consumeRequest.getMessageQueue());
}
break;
default:
break;
}
這再次說明，廣播模式是不支持消息重試的。
3、重平衡
在消費者啟動過程中，會調(diào)用 RebalanceService 的 start 方法，進行重平衡。從重平衡的代碼中可以看到，廣播模式消費者會消費所有 MessageQueue，而集群模式下會根據(jù)負載均衡策略選擇其中幾個 MessageQueue。代碼如下：
private void rebalanceByTopic(final String topic, final boolean isOrder) {
switch (messageModel) {
case BROADCASTING: {
Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
if (mqSet != null) {
boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder);
//省略部分邏輯
} else {
}
break;
}
case CLUSTERING: {
Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
//省略部分邏輯
if (mqSet != null && cidAll != null) {
//省略部分邏輯
try {
allocateResult = strategy.allocate(
this.consumerGroup,
this.mQClientFactory.getClientId(),
mqAll,
cidAll);
} catch (Throwable e) {
return;
}
Set<MessageQueue> allocateResultSet = new HashSet<MessageQueue>();
if (allocateResult != null) {
allocateResultSet.addAll(allocateResult);
}
boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet, isOrder);
//省略部分邏輯
}
break;
}
default:
break;
}
}
上面 updateProcessQueueTableInRebalance 這個方法調(diào)用前，要獲取到需要消費的 MessageQueue 集合。廣播模式下，直接取了訂閱的 Topic 下的所有集合元素，而集群模式下，則需要通過負責均衡獲取當前消費者自己要消費的 MessageQueue 集合。
4、總結(jié)
本文主要講解了 RocketMQ 廣播消息的實現(xiàn)機制，理解廣播消息，要把握下面幾點:
1.偏移量保存在消費者本地內(nèi)存和文件中。
2.廣播消息不支持重試。
3.從源碼上看，廣播模式并不能支持順序消息。
4.廣播模式消費者訂閱了 Topic 下的所有 MessageQueue，不會重平衡。
