29.RocketMQ之消费者负载均衡策略-Toy模板网

这篇具有很好参考价值的文章主要介绍了29.RocketMQ之消费者负载均衡策略。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

highlight: arduino-light

负载均衡策略概述

消费者在消费消息的时候，需要知道从Broker的哪一个消息队列中去获取消息。所以，在消费者端必须要做负载均衡，即Broker端中多个消费队列分配给同一个消费者组中的哪些消费者消费。

上篇文章讲解了负载均衡过程，这篇文章讲解消费者负载均衡的策略。

AllocateMessageQueueStrategy是负载均衡策略接口，有两个方法，分别是allocate和getName，allocate方法是为消费者分配消息队列，getName是获取负载均衡策略的名字。AllocateMessageQueueStrategy接口有六个子类实现。

负载均衡策略说明

它们具体说明如下：

AllocateMessageQueueAveragely：平均负载策略，将消息队列平均分配给每一个消息者。假设一个topic有8个消息队列，有3个消息者A、B、C，那么采用该负载策略，那么A首先分配3个消息队列，然后B也分配3个消息队列，最后C分配2个消息队列。
AllocateMessageQueueAveragelyByCircle：循环平均负载策略，跟平均负载策略不同的是。是将消息队列一个一个的分配给消息者。假设一个topic有8个消息队列，有3个消息者A、B、C。采用循环平均负载策略分配的方法是首先给A、B、C分别分配一个，然后再进行第二轮分配，也是给A、B、C分别再分配一个，这样子还剩下两个就分别分配给A、B。
AllocateMessageQueueByConfig：配置负载策略，用户直接给消费者分配配置消息队列。
AllocateMessageQueueConsistentHash：一致性哈希负载策略，为每一个消费者创建多个虚拟的节点，将虚拟节点连成一个环，这个环就是一致性哈希环，然后将消息队列进行哈希计算得到哈希值，通过哈希值找到距离一致性哈希环顺时针方向最近的那个虚拟节点，此时就可以通过虚拟节点获取到真实的消费者了，就将这个消息队列分配给这个消息者。
AllocateMessageQueueByMachineRoom：同机房分配策略，将Broker的消息队列分配给同机房的消费者。
AllocateMachineRoomNearby：AllocateMessageQueueByMachineRoom策略的升级版本，不仅将Broker的消息队列分配给同机房的消费者，还会将剩下的消息队列根据给定的分配策略进行分配给消费者。

接下来，将从源码的层面分析上述的负载均衡策略。

AllocateMessageQueueAveragely

```java //代码位置：org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely#allocate public List allocate(String consumerGroup, String currentCID, List mqAll, List cidAll) {

//省略校验合法性的代码

    //获取当前客户端id在所有消费者里面的位置
    int index = cidAll.indexOf(currentCID);
    //看是否能够均分
    int mod = mqAll.size() % cidAll.size();
    //如果消息队列的数量小于等于消费者的数量。一个消费者最多只能分到一个消息消息队列
    //否则，如果不能均分且index小于mode，均分以后再加上一个
    int averageSize =
        mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size()
            + 1 : mqAll.size() / cidAll.size());
    //开始的位置
    int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;
    //被分配的消息队列的范围上限
    int range = Math.min(averageSize, mqAll.size() - startIndex);
    for (int i = 0; i < range; i++) {
        result.add(mqAll.get((startIndex + i) % mqAll.size()));
    }
    return result;

} ```

AllocateMessageQueueAveragely的allocate方法的逻辑：

获取当前消费者在所有消费者里面的位置(索引)
计算当前消费者能分配的消息队列的数量，以及分配的消息队列的范围，即从哪里开始分配到哪里结束分配。
根据分配消息队列的范围给消费者进行分配。

AllocateMessageQueueAveragelyByCircle

```java //代码位置： org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragelyByCircle#allocate public List allocate(String consumerGroup, String currentCID, List mqAll, List cidAll) {

//省略校验合法性的代码

    //当前consumer排序后的索引
    int index = cidAll.indexOf(currentCID);
    for (int i = index; i < mqAll.size(); i++) {
        //取模
        if (i % cidAll.size() == index) {
            result.add(mqAll.get(i));
        }
    }
    return result;

} ```

AllocateMessageQueueAveragelyByCircle的allocate方法的逻辑：

获取当前消费者在所有消费者里面的位置index(索引)
遍历所有的消息队列，当前消息队列的索引对消费者数量进行取模的值等于index，将此此时的消息队列分配给消费者。

AllocateMessageQueueByConfig

```java /代码位置：org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueByConfig#allocate private List messageQueueList;

public void setMessageQueueList(List messageQueueList) { this.messageQueueList = messageQueueList; }

@Override

public List allocate(String consumerGroup, String currentCID, List mqAll, List cidAll) { return this.messageQueueList; } ```

AllocateMessageQueueByConfig类有一个messageQueueList属性，这个就是用户配置的消息队列。AllocateMessageQueueByConfig的allocate方法就是将用户配置的这个消息队列返回给消费者。

☆AllocateMessageQueueConsistentHash

```java //代码位置：org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueConsistentHash#allocate public List allocate(String consumerGroup, String currentCID, List mqAll, List cidAll) {

//省略校验合法性的代码

    //构建消费者节点
    Collection<ClientNode> cidNodes = new ArrayList<ClientNode>();
    for (String cid : cidAll) {
        cidNodes.add(new ClientNode(cid));
    }

    //用消费者节点生产多个虚拟的节点，构建哈希环
    final ConsistentHashRouter<ClientNode> router; //for building hash ring
    if (customHashFunction != null) {
        router = new ConsistentHashRouter<ClientNode>(cidNodes, virtualNodeCnt, customHashFunction);
    } else {
        router = new ConsistentHashRouter<ClientNode>(cidNodes, virtualNodeCnt);
    }


    List<MessageQueue> results = new ArrayList<MessageQueue>();
    //遍历消息队列
    for (MessageQueue mq : mqAll) {
        //对messageQueue进行hash计算，找到顺时针最近的consumer节点
        ClientNode clientNode = router.routeNode(mq.toString());
        //判断是否是当前consumer
        if (clientNode != null && currentCID.equals(clientNode.getKey())) {
            results.add(mq);
        }
    }

    return results;

} ```

AllocateMessageQueueConsistentHash的allocate方法逻辑：

构建消费者节点
用消费者节点生成多个虚拟的节点，构建哈希环
遍历消息队列，对消息队列进行hash计算，找到距离最近的消费者节点，将此时的消息队列分配给消费者。

AllocateMessageQueueConsistentHash的allocate方法实现了一致性哈希算法,如果后续有使用一致性哈希算法的场景可拿来参考。

AllocateMessageQueueByMachineRoom

```java //代码位置：org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueByMachineRoom#allocate public List allocate(String consumerGroup, String currentCID, List mqAll, //省略校验合法性的代码 List premqAll = new ArrayList (); for (MessageQueue mq : mqAll) { //brokerName命名规则 machine_room1@broker-a String[] temp = mq.getBrokerName().split("@"); //判断是否符合指定的机房条件 if (temp.length == 2 && consumeridcs.contains(temp[0])) { premqAll.add(mq); } }

//平均分配到的队列数
    int mod = premqAll.size() / cidAll.size();
    //取模
    int rem = premqAll.size() % cidAll.size();
    //当前分配到的第一个队列索引
    int startIndex = mod * currentIndex;
    //当前分配到的最后一个队列索引
    int endIndex = startIndex + mod;
    //取startIndex到endIndex的队列
    for (int i = startIndex; i < endIndex; i++) {
        result.add(mqAll.get(i));
    }
    //如果不能平均分配，并且模大于当前的索引，再分配一个
    if (rem > currentIndex) {
        result.add(premqAll.get(currentIndex + mod * cidAll.size()));
    }
    return result;

} ```

AllocateMessageQueueByMachineRoom的allocate方法逻辑：

遍历所有的消息队列，将所有的指定机房的消息队列收集起来。
将上述获取得到的消息队列平均分配给消费者，如果不能分配，再次分配一个消息队列。

AllocateMachineRoomNearby

```java //代码位置：org.apache.rocketmq.client.consumer.rebalance.AllocateMachineRoomNearby#allocate public List allocate(String consumerGroup, String currentCID, List mqAll, //省略校验合法性的代码

//group mq by machine room
    //根据机房对队列分组
    Map<String/*machine room */, List<MessageQueue>> mr2Mq = new TreeMap<String, List<MessageQueue>>();
    for (MessageQueue mq : mqAll) {
        //判断当前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);
        }
    }

    //group consumer by machine room
    //consumer按机房分组
    Map<String/*machine room */, List<String/*clientId*/>> mr2c = new TreeMap<String, List<String>>();
    for (String cid : cidAll) {
        //判断consumer处于哪个机房
        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);
        }
    }

    List<MessageQueue> allocateResults = new ArrayList<MessageQueue>();

    //1.allocate the mq that deploy in the same machine room with the current consumer
    //给当前consumer分当前机房的那些MessageQeueue
    String currentMachineRoom = machineRoomResolver.consumerDeployIn(currentCID);
    //得到当前机房的MessageQueue
    List<MessageQueue> mqInThisMachineRoom = mr2Mq.remove(currentMachineRoom);
    //得到当前机房的Consumer
    List<String> consumerInThisMachineRoom = mr2c.get(currentMachineRoom);
    if (mqInThisMachineRoom != null && !mqInThisMachineRoom.isEmpty()) {
        //得到当前机房所有MessageQueue和Consumers后根据指定的策略再负载
        allocateResults.addAll(allocateMessageQueueStrategy.allocate(consumerGroup, currentCID, mqInThisMachineRoom, consumerInThisMachineRoom));
    }

    //2.allocate the rest mq to each machine room if there are no consumer alive in that machine room
    //如果该MessageQueue的机房 没有同机房的consumer,将这些MessageQueue按配置好的备用策略分配给所有的consumer
    for (String machineRoom : mr2Mq.keySet()) {
        if (!mr2c.containsKey(machineRoom)) { // no alive consumer in the corresponding machine room, so all consumers share these queues
            //添加分配到的游离态MessageQueue
            allocateResults.addAll(allocateMessageQueueStrategy.allocate(consumerGroup, currentCID, mr2Mq.get(machineRoom), cidAll));
        }
    }

    return allocateResults;
}

```

AllocateMachineRoomNearby的allocate方法逻辑：