当前位置： 首页 > news >正文

RocketMQ 消费者Rebalance算法 解析——图解、源码级解析

news 来源：原创 2024/5/2 0:55:02

🍊 Java学习：Java从入门到精通总结

🍊 深入浅出RocketMQ设计思想：深入浅出RocketMQ设计思想

🍊 绝对不一样的职场干货：大厂最佳实践经验指南


📆 最近更新：2022年10月15日

🍊 个人简介：通信工程本硕💪、Java程序员🌕。做过科研paper，发过专利，优秀的程序员不应该只是CRUD

🍊 点赞 👍 收藏 ⭐留言 📝 都是我最大的动力！

文章目录

  • 平均分配算法
  • 环形平均分配算法
  • 一致性哈希算法
  • 指定机房算法
  • 就进机房算法
  • 手动配置负载均衡参数

平均分配算法

这也是消息消费时候的默认算法，所谓平均，就是同一个Topic主题下的所有队列被同一个消费者组中的所有Consumer平均消费掉。

例如有5个队列和2和Consumer，就会根据下面的步骤进行分配：

  1. 5除以2不能整除，所以队列无法均分
  2. 每个消费者先分到2个队列
  3. 多出来的1个队列按照顺序分配给了第一个Consumer

在这里插入图片描述
具体的源码如下：

public class AllocateMessageQueueAveragely implements AllocateMessageQueueStrategy {
    private final InternalLogger log = ClientLogger.getLog();

    @Override
    public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,
        List<String> cidAll) {
        if (currentCID == null || currentCID.length() < 1) {
            throw new IllegalArgumentException("currentCID is empty");
        }
        if (mqAll == null || mqAll.isEmpty()) {
            throw new IllegalArgumentException("mqAll is null or mqAll empty");
        }
        if (cidAll == null || cidAll.isEmpty()) {
            throw new IllegalArgumentException("cidAll is null or cidAll empty");
        }

        List<MessageQueue> result = new ArrayList<MessageQueue>();
        if (!cidAll.contains(currentCID)) {
            log.info("[BUG] ConsumerGroup: {} The consumerId: {} not in cidAll: {}",
                consumerGroup,
                currentCID,
                cidAll);
            return result;
        }
		// 当前分配到的Consumer的索引
        int index = cidAll.indexOf(currentCID);
        // 余数
        int mod = mqAll.size() % cidAll.size();
		
		// 队列总数小于Consumer总数时，给当前Consumer分配一个队列消费
		// 不能均分且当前编号小于余数时，需要给当前Consumer分配x + 1个队列，否则分配x个队列
        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;

		// 取min的原因是，如果Consumer多，队列少，多出来的Consumer分配不到队列
        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;
    }

    @Override
    public String getName() {
        return "AVG";
    }
}

环形平均分配算法

使用方法：

consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueAveragelyByCircle());

也可以自定义消费策略：

consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueStrategy() {
            @Override
            public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll, List<String> cidAll) {
                // 自定义负载策略
                return null;
            }

            @Override
            public String getName() {
                return null;
            }
        });

所谓环形分配算法，就是把消息队列按照环形进行排列，然后同一个组下的所有Consumer按照顺序进行匹配即可，如下图所示：
在这里插入图片描述
上图中Topic下共有10个消息队列，假设消费者组里有4个Consumer，分配过程如下：

  1. 对所有的消息队列和Consumer分别排序
  2. 按照顺序让Consumer和消息队列进行匹配

第一轮分配Queue1到Queue4，第二轮分配Queue5到Queue8，第三轮分配Queue9和Queue10。经过3轮分配完毕

具体源码如下所示：

public class AllocateMessageQueueAveragelyByCircle implements AllocateMessageQueueStrategy {
    private final InternalLogger log = ClientLogger.getLog();

    @Override
    public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,
        List<String> cidAll) {
        if (currentCID == null || currentCID.length() < 1) {
            throw new IllegalArgumentException("currentCID is empty");
        }
        if (mqAll == null || mqAll.isEmpty()) {
            throw new IllegalArgumentException("mqAll is null or mqAll empty");
        }
        if (cidAll == null || cidAll.isEmpty()) {
            throw new IllegalArgumentException("cidAll is null or cidAll empty");
        }

        List<MessageQueue> result = new ArrayList<MessageQueue>();
        if (!cidAll.contains(currentCID)) {
            log.info("[BUG] ConsumerGroup: {} The consumerId: {} not in cidAll: {}",
                consumerGroup,
                currentCID,
                cidAll);
            return result;
        }

        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;
    }

    @Override
    public String getName() {
        return "AVG_BY_CIRCLE";
    }
}



一致性哈希算法

首先先介绍一下一致性哈希算法：

hash算法带来的问题：
假设后台有多个服务器，我们就可以做负载均衡将前端来的请求“平均”分配到各个服务器上来处理，如果是按照用户id对服务器个数N取模来计算hash的话，如果有一台服务器宕机，之前所有的求模计算都要重来，开销较大。


一致性哈希的思路就是：用户按照顺时针方向做排列，离哪个节点近，就去访问哪个节点。
请添加图片描述
用户按照顺时针方向，离哪个节点近，就去访问哪个节点。

此时如果有服务器宕机，直接顺着找下一个服务器节点就可以了。
请添加图片描述如果要增加节点：
请添加图片描述

RocketMQ中对于一致性哈希的源码级实现：

public class AllocateMessageQueueConsistentHash implements AllocateMessageQueueStrategy {
    private final InternalLogger log = ClientLogger.getLog();

    private final int virtualNodeCnt;
    private final HashFunction customHashFunction;

    public AllocateMessageQueueConsistentHash() {
        this(10);
    }
	
	// 设计虚拟节点数量
    public AllocateMessageQueueConsistentHash(int virtualNodeCnt) {
        this(virtualNodeCnt, null);
    }

    public AllocateMessageQueueConsistentHash(int virtualNodeCnt, HashFunction customHashFunction) {
        if (virtualNodeCnt < 0) {
            throw new IllegalArgumentException("illegal virtualNodeCnt :" + virtualNodeCnt);
        }
        this.virtualNodeCnt = virtualNodeCnt;
        this.customHashFunction = customHashFunction;
    }
	
	// 负载均衡算法主要实现
    @Override
    public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,
        List<String> cidAll) {

        if (currentCID == null || currentCID.length() < 1) {
            throw new IllegalArgumentException("currentCID is empty");
        }
        if (mqAll == null || mqAll.isEmpty()) {
            throw new IllegalArgumentException("mqAll is null or mqAll empty");
        }
        if (cidAll == null || cidAll.isEmpty()) {
            throw new IllegalArgumentException("cidAll is null or cidAll empty");
        }

        List<MessageQueue> result = new ArrayList<MessageQueue>();
        if (!cidAll.contains(currentCID)) {
            log.info("[BUG] ConsumerGroup: {} The consumerId: {} not in cidAll: {}",
                consumerGroup,
                currentCID,
                cidAll);
            return result;
        }
		
		// 把所有消费者放到一个List里
        Collection<ClientNode> cidNodes = new ArrayList<ClientNode>();
        for (String cid : cidAll) {
            cidNodes.add(new ClientNode(cid));
        }
		
		// 创建hash环形结构
        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);
        }
		
		// 根据一致性hash算法，基于客户端节点，把分配到当前消费者组的MQ添加到集合里并返回
        List<MessageQueue> results = new ArrayList<MessageQueue>();
        for (MessageQueue mq : mqAll) {
            ClientNode clientNode = router.routeNode(mq.toString());
            if (clientNode != null && currentCID.equals(clientNode.getKey())) {
                results.add(mq);
            }
        }

        return results;

    }

    @Override
    public String getName() {
        return "CONSISTENT_HASH";
    }

    private static class ClientNode implements Node {
        private final String clientID;

        public ClientNode(String clientID) {
            this.clientID = clientID;
        }

        @Override
        public String getKey() {
            return clientID;
        }
    }

}

上面代码在ConsistentHashRouter中创建了hash环，算法的主要流程是在这个类中实现的，主要是基于TreeMap，感兴趣的小伙伴可以深入研究一下它的源码~



指定机房算法

假设有两个机房，则对应的消费关系如下图：
在这里插入图片描述
指定机房分配算法先根据MQ所述的Broker找出有效的机房里的所有MQ，然后再平分给所有的Consumer

public class AllocateMessageQueueByMachineRoom implements AllocateMessageQueueStrategy {
    private Set<String> consumeridcs;

    @Override
    public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,
        List<String> cidAll) {
        List<MessageQueue> result = new ArrayList<MessageQueue>();
        // 计算出当前消费者ID在消费者集合中的具体位置
        int currentIndex = cidAll.indexOf(currentCID);
        if (currentIndex < 0) {
            return result;
        }
        // 拿出BrokerName下的所有MQ
        List<MessageQueue> premqAll = new ArrayList<MessageQueue>();
        for (MessageQueue mq : mqAll) {
            String[] temp = mq.getBrokerName().split("@");
            if (temp.length == 2 && consumeridcs.contains(temp[0])) {
                premqAll.add(mq);
            }
        }
        
		// 队列长度除以客户端长度
        int mod = premqAll.size() / cidAll.size();
        // 队列长度mod客户端长度
        int rem = premqAll.size() % cidAll.size();
        
        // 给Consumer分配MQ
        int startIndex = mod * currentIndex;
        int endIndex = startIndex + mod;
        for (int i = startIndex; i < endIndex; i++) {
            result.add(premqAll.get(i));
        }
        if (rem > currentIndex) {
            result.add(premqAll.get(currentIndex + mod * cidAll.size()));
        }
        return result;
    }

    @Override
    public String getName() {
        return "MACHINE_ROOM";
    }

    public Set<String> getConsumeridcs() {
        return consumeridcs;
    }

    public void setConsumeridcs(Set<String> consumeridcs) {
        this.consumeridcs = consumeridcs;
    }
}



就进机房算法

顾名思义，就近机房分配策略是一种基于Consumer和机房距离来分配的策略。部署在同一个机房的MQ会被先分配给同一个机房里的Consumer。

具体步骤是先统计Consumer与Broker所在的机房，之后再将Broker中的MQ分配给同机房的Consumer消费，如果本机房里没有Consumer，则再尝试分配给其他机房的Consumer

在这里插入图片描述

public class AllocateMachineRoomNearby implements AllocateMessageQueueStrategy {
    private final InternalLogger log = ClientLogger.getLog();

    private final AllocateMessageQueueStrategy allocateMessageQueueStrategy;//actual allocate strategy
    private final MachineRoomResolver machineRoomResolver;

    public AllocateMachineRoomNearby(AllocateMessageQueueStrategy allocateMessageQueueStrategy,
        MachineRoomResolver machineRoomResolver) throws NullPointerException {
        if (allocateMessageQueueStrategy == null) {
            throw new NullPointerException("allocateMessageQueueStrategy is null");
        }

        if (machineRoomResolver == null) {
            throw new NullPointerException("machineRoomResolver is null");
        }

        this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
        this.machineRoomResolver = machineRoomResolver;
    }

    @Override
    public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,
        List<String> cidAll) {
        if (currentCID == null || currentCID.length() < 1) {
            throw new IllegalArgumentException("currentCID is empty");
        }
        if (mqAll == null || mqAll.isEmpty()) {
            throw new IllegalArgumentException("mqAll is null or mqAll empty");
        }
        if (cidAll == null || cidAll.isEmpty()) {
            throw new IllegalArgumentException("cidAll is null or cidAll empty");
        }

        List<MessageQueue> result = new ArrayList<MessageQueue>();
        if (!cidAll.contains(currentCID)) {
            log.info("[BUG] ConsumerGroup: {} The consumerId: {} not in cidAll: {}",
                consumerGroup,
                currentCID,
                cidAll);
            return result;
        }

        // 将MQ按照不同的机房归纳
        Map<String/*machine room */, List<MessageQueue>> mr2Mq = new TreeMap<String, List<MessageQueue>>();
        for (MessageQueue mq : mqAll) {
            String brokerMachineRoom = machineRoomResolver.brokerDeployIn(mq);
            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) {
            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. 分配与当前消费者部署在同一机房的MQ
        String currentMachineRoom = machineRoomResolver.consumerDeployIn(currentCID);
        List<MessageQueue> mqInThisMachineRoom = mr2Mq.remove(currentMachineRoom);
        List<String> consumerInThisMachineRoom = mr2c.get(currentMachineRoom);
        if (mqInThisMachineRoom != null && !mqInThisMachineRoom.isEmpty()) {
            allocateResults.addAll(allocateMessageQueueStrategy.allocate(consumerGroup, currentCID, mqInThisMachineRoom, consumerInThisMachineRoom));
        }

        //2.如果机房没有活着的消费者，则将其MQ分配给每个其他的机房
        for (String machineRoom : mr2Mq.keySet()) {
            if (!mr2c.containsKey(machineRoom)) { // no alive consumer in the corresponding machine room, so all consumers share these queues
                allocateResults.addAll(allocateMessageQueueStrategy.allocate(consumerGroup, currentCID, mr2Mq.get(machineRoom), cidAll));
            }
        }

        return allocateResults;
    }

    @Override
    public String getName() {
        return "MACHINE_ROOM_NEARBY" + "-" + allocateMessageQueueStrategy.getName();
    }

    /**
     * 一个解析器对象，用于确定消息队列或客户端部署在哪个机房。
     *
     * AllocateMachineRoomNearby将使用该结果按机房对消息队列和客户端进行分组。
     *
     * 返回值不能为null
     */
    public interface MachineRoomResolver {
        String brokerDeployIn(MessageQueue messageQueue);

        String consumerDeployIn(String clientID);
    }
}



手动配置负载均衡参数

除了使用内置的负载均衡算法以外，还可以手动配置相关的参数，例如设置消费的队列、消费的Topic、消费的机器等，在消费端直接设置消费队列即可：

consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueByConfig(){
            {
                this.setMessageQueueList(Collections.<MessageQueue>singletonList(new MessageQueue(){{
                    this.setQueueId(0);
                    this.setTopic("Topic name");
                    this.setBrokerName("Broker name");
                }}));
            }
        });

上面的代码里，手动指定了消费队列的索引，Topic和Broker服务器的名称，之后Consumer就会在指定的服务器中进行消费，源码如下：

public class AllocateMessageQueueByConfig implements AllocateMessageQueueStrategy {
    private List<MessageQueue> messageQueueList;

    @Override
    public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,
        List<String> cidAll) {
        return this.messageQueueList;
    }

    @Override
    public String getName() {
        return "CONFIG";
    }

    public List<MessageQueue> getMessageQueueList() {
        return messageQueueList;
    }

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

可以看到源码里只提供了一个消息队列集合，就是我们上面传入的自定义配置的MQ列表，配置完成之后就可以进行负载均衡及消费。

相关文章：

  • 【网络安全】内网杀器之Cobalt Strike
  • 法国1油井反杀苏军2油井经典案例分析
  • 【一年总结】我的大三
  • 【TS系列】TypeScript进阶（三）
  • 动态内存管理
  • Linux05开机/重启/用户登录注销
  • MySQL索引-2聚集索引探讨
  • 【node.js从入门到精通】模块化+npm包详解
  • Android 面试需要哪些准备？
  • Vue进阶--render函数和jsx的使用
  • 嵌入式 Linux 入门（十、Linux 下的 C 编程）
  • Spring boot 使用QQ邮箱进行一个验证登入
  • ＜Linux系统复习＞命令行参数和环境变量
  • Java框架详解1——Spring的设计思想
  • 【C++】类和对象 (下篇)
  • Apache Zeppelin在Apache Trafodion上的可视化
  • CSS3 变换
  • Druid 在有赞的实践
  • Dubbo 整合 Pinpoint 做分布式服务请求跟踪
  • JavaScript 基本功--面试宝典
  • Java知识点总结（JavaIO-打印流）
  • java中的hashCode
  • jquery cookie
  • Zepto.js源码学习之二
  • 关于Flux,Vuex,Redux的思考
  • 基于Volley网络库实现加载多种网络图片（包括GIF动态图片、圆形图片、普通图片）...
  • 蓝海存储开关机注意事项总结
  • 利用阿里云 OSS 搭建私有 Docker 仓库
  • 聊聊redis的数据结构的应用
  • 如何打造100亿SDK累计覆盖量的大数据系统
  • 体验javascript之美-第五课 匿名函数自执行和闭包是一回事儿吗？
  • 微信端页面使用-webkit-box和绝对定位时，元素上移的问题
  • 赢得Docker挑战最佳实践
  • 这几个编码小技巧将令你 PHP 代码更加简洁
  • TPG领衔财团投资轻奢珠宝品牌APM Monaco
  • #162 (Div. 2)
  • #define 用法
  • #ubuntu# #git# repository git config --global --add safe.directory
  • #设计模式#4.6 Flyweight（享元） 对象结构型模式
  • （20050108）又读《平凡的世界》
  • （C语言）fread与fwrite详解
  • （k8s中）docker netty OOM问题记录
  • （二）正点原子I.MX6ULL u-boot移植
  • (附源码)计算机毕业设计SSM智能化管理的仓库管理
  • （机器学习的矩阵）（向量、矩阵与多元线性回归）
  • （机器学习-深度学习快速入门）第一章第一节：Python环境和数据分析
  • (生成器)yield与(迭代器)generator
  • （小白学Java）Java简介和基本配置
  • （转）C#开发微信门户及应用(1)--开始使用微信接口
  • (转)C语言家族扩展收藏 (转)C语言家族扩展
  • .NET Compact Framework 3.5 支持 WCF 的子集
  • .NET微信公众号开发-2.0创建自定义菜单
  • .vue文件怎么使用_vue调试工具vue-devtools的安装
  • @LoadBalanced 和 @RefreshScope 同时使用，负载均衡失效分析
  • []sim300 GPRS数据收发程序