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

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

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("testGroup");
consumer.setNamesrvAddr("127.0.0.1:9876");
//訂閱topic
consumer.subscribe("testMsg","*");
//注冊消息監(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);
//啟動consumer
consumer.start();

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("testGroup");
consumer.setNamesrvAddr("127.0.0.1:9876");
//訂閱topic
consumer.subscribe("testMsg","*");
//注冊消息監(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();
//指定機房名稱  machine_room1、machine_room2
allocateMachineRoom.setConsumeridcs(new HashSet<>(Arrays.asList("machine_room1","machine_room2")));
//設(shè)置consumer的負(fù)載策略
consumer.setAllocateMessageQueueStrategy(allocateMachineRoom);
//啟動consumer
consumer.start();

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

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

//消息隊列按機房分組
Map<String/*machine room */, List<MessageQueue>> mr2Mq = new TreeMap<String, List<MessageQueue>>();
for (MessageQueue mq : mqAll) {
  //這里調(diào)用我們自己定義的類方法，得到broker的機房的名稱
  String brokerMachineRoom = machineRoomResolver.brokerDeployIn(mq);
  //機房不為空，將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按機房分組
Map<String/*machine room */, List<String/*clientId*/>> mr2c = new TreeMap<String, List<String>>();
for (String cid : cidAll) {
  //這里調(diào)用我們自己定義的類方法，得到broker的機房的名稱
  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)前機房的那些MessageQeueue
String currentMachineRoom = machineRoomResolver.consumerDeployIn(currentCID);
//得到當(dāng)前機房的MessageQueue
List<MessageQueue> mqInThisMachineRoom = mr2Mq.remove(currentMachineRoom);
//得到當(dāng)前機房的Consumer
List<String> consumerInThisMachineRoom = mr2c.get(currentMachineRoom);
if (mqInThisMachineRoom != null && !mqInThisMachineRoom.isEmpty()) {
  //得到當(dāng)前機房所有MessageQueue和Consumers后根據(jù)指定的策略再負(fù)載
  allocateResults.addAll(allocateMessageQueueStrategy.allocate(consumerGroup, currentCID, mqInThisMachineRoom, consumerInThisMachineRoom));
}
//2.如果該MessageQueue的機房 沒有同機房的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));
  }
}

//將所有consumer變成節(jié)點 到時候經(jīng)過hash計算 分布在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計算
  router = new ConsistentHashRouter<ClientNode>(cidNodes, virtualNodeCnt);
}
List<MessageQueue> results = new ArrayList<MessageQueue>();
for (MessageQueue mq : mqAll) {
  //對messageQueue進(jìn)行hash計算，找到順時針最近的consumer節(jié)點
  ClientNode clientNode = router.routeNode(mq.toString());
  //判斷是否是當(dāng)前consumer
  if (clientNode != null && currentCID.equals(clientNode.getKey())) {
    results.add(mq);
  }
}