前言

首先明確一點(diǎn)，RocketMQ 是基于發(fā)布訂閱模型的消息中間件。所謂的發(fā)布訂閱就是說，consumer 訂閱了 broker 上的某個(gè) topic，當(dāng) producer 發(fā)布消息到 broker 上的該 topic 時(shí)，consumer 就能收到該條消息。

之前我們講過 consumer group 的概念，即消費(fèi)同一類消息的多個(gè) consumer 實(shí)例組成一個(gè)消費(fèi)者組，也可以稱為一個(gè) consumer 集群，這些 consumer 實(shí)例使用同一個(gè) group name。需要注意一點(diǎn)，除了使用同一個(gè) group name，訂閱的 tag 也必須是一樣的，只有符合這兩個(gè)條件的 consumer 實(shí)例才能組成 consumer 集群。

1. 消費(fèi)模式

1.1 集群消費(fèi)

當(dāng) consumer 使用集群消費(fèi)時(shí)，每條消息只會(huì)被 consumer 集群內(nèi)的任意一個(gè) consumer 實(shí)例消費(fèi)一次。舉個(gè)例子，當(dāng)一個(gè) consumer 集群內(nèi)有 3 個(gè)consumer 實(shí)例（假設(shè)為consumer 1、consumer 2、consumer 3）時(shí)，一條消息投遞過來，只會(huì)被consumer 1、consumer 2、consumer 3中的一個(gè)消費(fèi)。

同時(shí)記住一點(diǎn)，使用集群消費(fèi)的時(shí)候，consumer 的消費(fèi)進(jìn)度是存儲(chǔ)在 broker 上，consumer 自身是不存儲(chǔ)消費(fèi)進(jìn)度的。消息進(jìn)度存儲(chǔ)在 broker 上的好處在于，當(dāng)你 consumer 集群是擴(kuò)大或者縮小時(shí)，由于消費(fèi)進(jìn)度統(tǒng)一在broker上，消息重復(fù)的概率會(huì)被大大降低了。

注意：在集群消費(fèi)模式下，并不能保證每一次消息失敗重投都投遞到同一個(gè) consumer 實(shí)例。

1.2 廣播消費(fèi)

當(dāng) consumer 使用廣播消費(fèi)時(shí)，每條消息都會(huì)被 consumer 集群內(nèi)所有的 consumer 實(shí)例消費(fèi)一次，也就是說每條消息至少被每一個(gè) consumer 實(shí)例消費(fèi)一次。舉個(gè)例子，當(dāng)一個(gè) consumer 集群內(nèi)有 3 個(gè) consumer 實(shí)例（假設(shè)為 consumer 1、consumer 2、consumer 3）時(shí)，一條消息投遞過來，會(huì)被 consumer 1、consumer 2、consumer 3都消費(fèi)一次。

與集群消費(fèi)不同的是，consumer 的消費(fèi)進(jìn)度是存儲(chǔ)在各個(gè) consumer 實(shí)例上，這就容易造成消息重復(fù)。還有很重要的一點(diǎn)，對(duì)于廣播消費(fèi)來說，是不會(huì)進(jìn)行消費(fèi)失敗重投的，所以在 consumer 端消費(fèi)邏輯處理時(shí)，需要額外關(guān)注消費(fèi)失敗的情況。

雖然廣播消費(fèi)能保證集群內(nèi)每個(gè) consumer 實(shí)例都能消費(fèi)消息，但是消費(fèi)進(jìn)度的維護(hù)、不具備消息重投的機(jī)制大大影響了實(shí)際的使用。因此，在實(shí)際使用中，更推薦使用集群消費(fèi)，因?yàn)榧合M(fèi)不僅擁有消費(fèi)進(jìn)度存儲(chǔ)的可靠性，還具有消息重投的機(jī)制。而且，我們通過集群消費(fèi)也可以達(dá)到廣播消費(fèi)的效果。

1.3 使用集群消費(fèi)模擬廣播消費(fèi)

如果業(yè)務(wù)上確實(shí)需要使用廣播消費(fèi)，那么我們可以通過創(chuàng)建多個(gè) consumer 實(shí)例，每個(gè) consumer 實(shí)例屬于不同的 consumer group，但是它們都訂閱同一個(gè) topic。舉個(gè)例子，我們創(chuàng)建 3 個(gè) consumer 實(shí)例，consumer 1（屬于consumer group 1）、consumer 2（屬于 consumer group 2）、consumer 3（屬于consumer group 3），它們都訂閱了 topic A ，那么當(dāng) producer 發(fā)送一條消息到 topic A 上時(shí)，由于 3 個(gè)consumer 屬于不同的 consumer group，所以 3 個(gè)consumer都能收到消息，也就達(dá)到了廣播消費(fèi)的效果了。 除此之外，每個(gè) consumer 實(shí)例的消費(fèi)邏輯可以一樣也可以不一樣，每個(gè)consumer group還可以根據(jù)需要增加 consumer 實(shí)例，比起廣播消費(fèi)來說更加靈活。

2. 消費(fèi)策略

Consumer在拉取消息之前需要對(duì)TopicMessage進(jìn)行負(fù)載操作，負(fù)載操作由一個(gè)定時(shí)器來完成單位，定時(shí)間隔默認(rèn)20s

簡(jiǎn)單來說就是將Topic下的MessageQueue分配給這些Consumer，至于怎么分，就是通過這些負(fù)載策略定義的算法規(guī)則來劃分。

2.1 AllocateMessageQueueAveragely

平均負(fù)載策略，RocketMQ默認(rèn)使用的就是這種方式，如果某個(gè)Consumer集群，訂閱了某個(gè)Topic，Topic下面的這些MessageQueue會(huì)被平均分配給集群中的Consumer，為了幫助大家理解，我畫了個(gè)圖

我來講下這個(gè)圖代表的意思，假設(shè)topic為 testMsg，該testMsg下面有4個(gè)MessageQueue，然后這些Consumer組成了一個(gè)集群（都監(jiān)聽了該Topic并且消費(fèi)groupId是一樣的），首先會(huì)給Consumer和MessageQueue進(jìn)行排序，誰是老大，誰先拿MessageQueue， 平均分配分為兩種情況

2.1.1 MessageQueue數(shù)量大于Consumer數(shù)量

如果隊(duì)列數(shù)量不是消費(fèi)者數(shù)量的整數(shù)倍，跟上圖中2個(gè)Consumer和4個(gè)Consumer的情況一樣，先分每個(gè)Consumer應(yīng)得的數(shù)量，拿2個(gè)Consumer來舉個(gè)例子，C1 和C2 各自分到了2個(gè)MessageQueue，C1排序時(shí)在C2前面，所以C1先把 Q0 和Q1拿走，C2再拿2個(gè)，也就是Q2和Q3。

如果隊(duì)列數(shù)量不是消費(fèi)者的整數(shù)倍，跟上圖3個(gè)Consumer和5個(gè)Consumer的情況一樣，5個(gè)Comsumer的比較特殊，我們過會(huì)再講，我們拿3個(gè)Consumer的情況來舉例，4個(gè)消息隊(duì)列，每個(gè)Consumer能分到1個(gè)，還剩下1個(gè)，弱肉強(qiáng)食嘛，剩下的當(dāng)然給排在前面的大哥C1啦，最后分下來，C1分到2個(gè)隊(duì)列，C2和C3只分到一個(gè)，分完數(shù)量以后，也是按照順序來拿，C1拿到了Q0和Q1，然后C2就只能從Q2開始拿，C3只能拿剩下的Q3啦

2.1.2 MessageQueue數(shù)量小于Consumer數(shù)量

這種情況平均下來，每個(gè)人1個(gè)Consumer都分不到一個(gè)，也就是我們上圖中的5個(gè)Comsumer的情況老規(guī)矩，按照排序順序，每人先拿1個(gè)隊(duì)列，由于C5排在最后面，隊(duì)列全被別人拿走了，C5就一直分不到消息隊(duì)列，除非前面的某個(gè)Consumer掛了，20s之后，在隊(duì)列重新負(fù)載的時(shí)候就能拿到MessageQueue。

具體算法：

//consumer的排序后的
int index = cidAll.indexOf(currentCID);
//取模
int mod = mqAll.size() % cidAll.size();
//如果隊(duì)列數(shù)小于消費(fèi)者數(shù)量，則將分到隊(duì)列數(shù)設(shè)置為1，如果余數(shù)大于當(dāng)前消費(fèi)者的index,則
//能分到的隊(duì)列數(shù)+1，否則就是平均值
int averageSize =
  mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size()
                                       + 1 : mqAll.size() / cidAll.size());
//consumer獲取第一個(gè)MessageQueue的索引
int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;
// 如果消費(fèi)者大于隊(duì)列數(shù)，rang會(huì)是負(fù)數(shù)，循環(huán)也就不會(huì)執(zhí)行 
int range = Math.min(averageSize, mqAll.size() - startIndex);
for (int i = 0; i < range; i++) {
  result.add(mqAll.get((startIndex + i) % mqAll.size()));
}

2.2 AllocateMessageQueueAveragelyByCircle

環(huán)形平均分配，這個(gè)和平均分配唯一的區(qū)別就是，再分隊(duì)列的時(shí)候，平均隊(duì)列是將屬于自己的MessageQueue全部拿走，而環(huán)形平均則是，一人拿一個(gè)，拿到的Queue不是連續(xù)的。我也畫了張圖來幫助大家理解

這種環(huán)形平均分配和平均分配，每個(gè)Consumer拿到的MessageQueue數(shù)量是不變的，我們就拿3個(gè)Consumer的情況舉個(gè)例子，

也是對(duì)Consumer和MessageQueue排序，先確定每個(gè)Consumer能拿到的MessageQueue數(shù)量，C1能分到2個(gè)，C2和C3只能分到1個(gè)

C1先拿1個(gè)，然后C2拿一個(gè)，C3拿一個(gè),C1再拿一個(gè)。也就是圖上3個(gè)Consumer畫的這個(gè)情況。

另外，如果Consumer的數(shù)量大于消息隊(duì)列的數(shù)量，處理方式和平均分配時(shí)一樣的。

//當(dāng)前consumer排序后的索引 
int index = cidAll.indexOf(currentCID);  
//index會(huì)是consumer第一個(gè)拿到的消息隊(duì)列索引
for (int i = index; i < mqAll.size(); i++) {
  //這里采用了取模的方式
  if (i % cidAll.size() == index) { 
    result.add(mqAll.get(i));  
  }
}

2.3 AllocateMessageQueueByConfig

用戶自定義配置，用戶在創(chuàng)建Consumer的時(shí)候，可以設(shè)置要使用的負(fù)載策略，如果我們?cè)O(shè)置為AllocateMessageQueueByConfig方式時(shí)，我們可以自己指定需要監(jiān)聽的MessageQueues，它維護(hù)了一個(gè)List messageQueueList，我們可以往這里面塞目標(biāo)的MessageQueues，這個(gè)策略了解一下就行，用的不多，具體設(shè)置代碼如下：

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("testGroup");
consumer.setNamesrvAddr("127.0.0.1:9876");
//訂閱topic
consumer.subscribe("testMsg","*");
//注冊(cè)消息監(jiān)聽
consumer.registerMessageListener(new MessageListenerConcurrently() {
  @Override
  public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
    // do job
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  }
});
//用戶自定義queue策略
AllocateMessageQueueByConfig allocateMessageQueueByConfig = new AllocateMessageQueueByConfig();
//指定MessageQueue
allocateMessageQueueByConfig.setMessageQueueList(Arrays.asList(new MessageQueue("testMsg","broker-a",0)));
//設(shè)置consumer的負(fù)載策略
consumer.setAllocateMessageQueueStrategy(allocateMessageQueueByConfig);
//啟動(dòng)consumer
consumer.start();

2.4 AllocateMessageQueueByMachineRoom

? 機(jī)房負(fù)載策略，其實(shí)這個(gè)策略就是當(dāng)前Consumer只負(fù)載處在指定的機(jī)房內(nèi)的MessageQueue，還有brokerName的命名必須要按要求的格式來設(shè)置：  機(jī)房名@brokerName

我們先看下具體的使用

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("testGroup");
consumer.setNamesrvAddr("127.0.0.1:9876");
//訂閱topic
consumer.subscribe("testMsg","*");
//注冊(cè)消息監(jiān)聽
consumer.registerMessageListener(new MessageListenerConcurrently() {
  @Override
  public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
    // do job
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  }
});
AllocateMessageQueueByMachineRoom allocateMachineRoom = new AllocateMessageQueueByMachineRoom();
//指定機(jī)房名稱  machine_room1、machine_room2
allocateMachineRoom.setConsumeridcs(new HashSet<>(Arrays.asList("machine_room1","machine_room2")));
//設(shè)置consumer的負(fù)載策略
consumer.setAllocateMessageQueueStrategy(allocateMachineRoom);
//啟動(dòng)consumer
consumer.start();

總結(jié)一下源碼的意思：

• 首先賽選出當(dāng)前Topic處在指定機(jī)房的隊(duì)列
• 賽選出隊(duì)列后，按照平均負(fù)載策略進(jìn)行具體的分配（算法極其相似）

哈哈，發(fā)現(xiàn)總結(jié)的真簡(jiǎn)潔呀，別慌，來張圖給你潤潤喉

其實(shí)這個(gè)策略就是對(duì)MessageQueue進(jìn)行了過濾，過濾完了以后，后續(xù)操作就按照平均負(fù)載策略來進(jìn)行具體負(fù)載操作。這個(gè)算法和平均負(fù)載的算法得到的結(jié)果是一樣的，我感覺這兩個(gè)策略應(yīng)該是兩個(gè)人寫的，不然不會(huì)寫兩套不同的算法來實(shí)現(xiàn)一個(gè)功能。也不知道是不是我自己太差了，理解不到大佬的思路。

2.5 AllocateMachineRoomNearby

這個(gè)策略我個(gè)人感覺是AllocateMessageQueueByMachineRoom的改進(jìn)版本，因?yàn)檫@個(gè)策略的處理方式要比AllocateMessageQueueByMachineRoom更加靈活，還考慮到了那些同機(jī)房只有MessageQueue卻沒有Consumer的情況，下面我們來具體講這個(gè)策略。使用該策略需要自己定義一個(gè)類，來區(qū)分每個(gè)broker處于哪個(gè)機(jī)房，該策略RocketMQ有個(gè)測(cè)試單元，我稍微改造了一下，就是把這個(gè)類提出來了。

public class MyMachineResolver implements AllocateMachineRoomNearby.MachineRoomResolver {
    /**
     * 判斷當(dāng)前broker處于哪個(gè)機(jī)房
     * @param messageQueue
     * @return
     */
    @Override
    public String brokerDeployIn(MessageQueue messageQueue) {
        return messageQueue.getBrokerName().split("-")[0];
    }
    /**
     * 判斷consumer處于哪個(gè)機(jī)房
     * @param clientID
     * @return
     */
    @Override
    public String consumerDeployIn(String clientID) {
        return clientID.split("-")[0];
    }
}

我們從代碼中，可以看出來需要在設(shè)置brokerName和Consumer的Id的時(shí)候需要加上機(jī)房名稱，eg：hz_aliyun_room1-broker-a、hz_aliyun_root1-Client1。我們先看一下在代碼里面怎么使用這個(gè)同機(jī)房分配策略

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("testGroup");
consumer.setNamesrvAddr("127.0.0.1:9876");
//訂閱topic
consumer.subscribe("testMsg","*");
//注冊(cè)消息監(jiān)聽
consumer.registerMessageListener(new MessageListenerConcurrently() {
  @Override
  public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
    // do job
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  }
});
//用戶同機(jī)房分配策略  
consumer.setAllocateMessageQueueStrategy(new AllocateMachineRoomNearby(new AllocateMessageQueueAveragely()
                ,new MyMachineResolver()));
//啟動(dòng)consumer
consumer.start();

我們可以看到，我們創(chuàng)建這個(gè)同機(jī)房分配策略的時(shí)候，還加了一個(gè)平均分配的策略進(jìn)去，它本身就是一個(gè)策略，為啥還要傳另一個(gè)策略。（該策略只會(huì)講MessageQueue和Consumer按機(jī)房進(jìn)行分組，分組以后具體的負(fù)載，就是通過我們傳的另外一個(gè)負(fù)載策略來分配的）我們到源碼里面去看，后面會(huì)解釋到

//消息隊(duì)列按機(jī)房分組
Map<String/*machine room */, List<MessageQueue>> mr2Mq = new TreeMap<String, List<MessageQueue>>();
for (MessageQueue mq : mqAll) {
  //這里調(diào)用我們自己定義的類方法，得到broker的機(jī)房的名稱
  String brokerMachineRoom = machineRoomResolver.brokerDeployIn(mq);
  //機(jī)房不為空，將broker放到分組中
  if (StringUtils.isNoneEmpty(brokerMachineRoom)) {
    if (mr2Mq.get(brokerMachineRoom) == null) {
      mr2Mq.put(brokerMachineRoom, new ArrayList<MessageQueue>());
    }
    mr2Mq.get(brokerMachineRoom).add(mq);
  } else {
    throw new IllegalArgumentException("Machine room is null for mq " + mq);
  }
}
//consumer按機(jī)房分組
Map<String/*machine room */, List<String/*clientId*/>> mr2c = new TreeMap<String, List<String>>();
for (String cid : cidAll) {
  //這里調(diào)用我們自己定義的類方法，得到broker的機(jī)房的名稱
  String consumerMachineRoom = machineRoomResolver.consumerDeployIn(cid);
  if (StringUtils.isNoneEmpty(consumerMachineRoom)) {
    if (mr2c.get(consumerMachineRoom) == null) {
      mr2c.put(consumerMachineRoom, new ArrayList<String>());
    }
    mr2c.get(consumerMachineRoom).add(cid);
  } else {
    throw new IllegalArgumentException("Machine room is null for consumer id " + cid);
  }
}
//當(dāng)前consumer分到的所有MessageQueue
List<MessageQueue> allocateResults = new ArrayList<MessageQueue>();
//1.給當(dāng)前consumer分當(dāng)前機(jī)房的那些MessageQeueue
String currentMachineRoom = machineRoomResolver.consumerDeployIn(currentCID);
//得到當(dāng)前機(jī)房的MessageQueue
List<MessageQueue> mqInThisMachineRoom = mr2Mq.remove(currentMachineRoom);
//得到當(dāng)前機(jī)房的Consumer
List<String> consumerInThisMachineRoom = mr2c.get(currentMachineRoom);
if (mqInThisMachineRoom != null && !mqInThisMachineRoom.isEmpty()) {
  //得到當(dāng)前機(jī)房所有MessageQueue和Consumers后根據(jù)指定的策略再負(fù)載
  allocateResults.addAll(allocateMessageQueueStrategy.allocate(consumerGroup, currentCID, mqInThisMachineRoom, consumerInThisMachineRoom));
}
//2.如果該MessageQueue的機(jī)房 沒有同機(jī)房的consumer,將這些MessageQueue按配置好的備用策略分配給所有的consumer
for (String machineRoom : mr2Mq.keySet()) {
  if (!mr2c.containsKey(machineRoom)) { 
    //添加分配到的游離態(tài)MessageQueue
    allocateResults.addAll(allocateMessageQueueStrategy.allocate(consumerGroup, currentCID, mr2Mq.get(machineRoom), cidAll));
  }
}

總結(jié)一下源碼的意思

1. 分別給MessageQueue和Consumer按機(jī)房分組
2. 得到當(dāng)前Consumer所在機(jī)房的所有Consumer和MessageQueue
3. 通過設(shè)置的負(fù)載策略，再進(jìn)行具體的負(fù)載，得到當(dāng)前Consumer分到的MessageQueue
4. 如果存在MessageQueue的某個(gè)機(jī)房中，沒有和MessageQueue同機(jī)房的Consumer，將這些MessageQueue按配置的負(fù)載策略分配給集群中所有的Consumer去負(fù)載
5. 最終該Consumer分到的MessageQueue會(huì)包含同機(jī)房分配到的和部分游離態(tài)分配的

這里我也畫個(gè)圖來解釋一下，

先同機(jī)房的Consumer和MessageQueue進(jìn)行負(fù)載，這里按照平均負(fù)載來分（我們創(chuàng)建機(jī)房就近策略使用的是平均負(fù)載），然后將游離態(tài)的通過設(shè)置的負(fù)載策略來分。

2.6 AllocateMessageQueueConsistentHash

一致性哈希策略，這里我簡(jiǎn)單介紹一下一致性哈希，這里不好我先給個(gè)圖，我們?cè)賮斫忉?/p>

一致性哈希有一個(gè)哈希環(huán)的概念，哈希環(huán)由數(shù)值 0到2^32-1 組成，不管內(nèi)容多長的字符，經(jīng)過哈希計(jì)算都能得到一個(gè)等長的數(shù)字，最后都會(huì)落在哈希環(huán)上的某個(gè)點(diǎn)，哈希環(huán)上的點(diǎn)都是虛擬的，比如我們這里使用Consumer的Id來進(jìn)行哈希計(jì)算，得到的這幾個(gè)是物理的點(diǎn)，然后把得到的點(diǎn)存到TreeMap里面，然后將所有的MessageQueue依次進(jìn)行同樣的哈希計(jì)算，得到距離MessageQueue順時(shí)針方向最近的那個(gè)Consumer點(diǎn)，這個(gè)就是MessageQeueu最終歸屬的那個(gè)Consumer。

我們看下源碼：

//將所有consumer變成節(jié)點(diǎn) 到時(shí)候經(jīng)過hash計(jì)算 分布在hash環(huán)上
Collection<ClientNode> cidNodes = new ArrayList<ClientNode>();
for (String cid : cidAll) {
  cidNodes.add(new ClientNode(cid));
}
final ConsistentHashRouter<ClientNode> router; 
//構(gòu)建哈希環(huán)
if (customHashFunction != null) {
  router = new ConsistentHashRouter<ClientNode>(cidNodes, virtualNodeCnt, customHashFunction);
} else {
  //默認(rèn)使用MD5進(jìn)行Hash計(jì)算
  router = new ConsistentHashRouter<ClientNode>(cidNodes, virtualNodeCnt);
}
List<MessageQueue> results = new ArrayList<MessageQueue>();
for (MessageQueue mq : mqAll) {
  //對(duì)messageQueue進(jìn)行hash計(jì)算，找到順時(shí)針最近的consumer節(jié)點(diǎn)
  ClientNode clientNode = router.routeNode(mq.toString());
  //判斷是否是當(dāng)前consumer
  if (clientNode != null && currentCID.equals(clientNode.getKey())) {
    results.add(mq);
  }
}

到此這篇關(guān)于RocketMQ中的消費(fèi)模式和消費(fèi)策略詳解的文章就介紹到這了,更多相關(guān)RocketMQ消費(fèi)模式內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論