RocketMQ源码(三)简单探索Producer和Consumer与Queue之间的负载均衡策略

本文涉及的产品
云服务器 ECS,每月免费额度200元 3个月
云服务器ECS,u1 2核4GB 1个月
简介: - Producer如何将消息负载均衡发送给queue?- Consumer如何通过负载均衡并发消费queue的消息?

RocketMQ源码(三)简单探索Producer和Consumer与Queue之间的负载均衡策略

RocketMQ架构中,我们都知道一个topic下可以创建多个queue,生产者通过负载均衡策略可以将消息均匀的分发在各个queue中,而这些queue
可以通过负载均衡给多个消费者订阅从而提升消费效率,本文将从以下两个方面从源码角度分析producer和consumer的负载均衡原理:

  • Producer如何将消息负载均衡发送给queue?
  • Consumer如何通过负载均衡并发消费queue的消息?

Tip以下是本人经过多年的工作经验集成的JavaWeb脚手架,封装了各种通用的starter可开箱即用,同时列举了互联网各种高性能场景的使用示例。

// Git代码
https://gitee.com/yeeevip/yeee-memo
https://github.com/yeeevip/yeee-memo

1 Producer下的负载均衡

在Producer下主要指的是通过负载均衡算法去选择一个queue去发送消息,这里默认使用的策略是轮询的方式按顺序选择Queue。

20231226-01.png

1.1 源码分析

  • 1 在没有指定Queue发送消息时,会调用到MQFaultStrategy.selectOneMessageQueue方法
  • 2 selectOneMessageQueue中调用tpInfo.selectOneMessageQueue()
  • 3 进而使用轮询算法从该Topic下的所有Queue里选择一个queue去发送消息

1.1.1 核心代码

public class MQFaultStrategy {
   
   
    public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName, final boolean resetIndex) {
   
   
        ...
        ...
        return tpInfo.selectOneMessageQueue();
    }
}
/*
    轮询算法:维护一个全局的index,每次都+1再与Queue的size求余
 */
public class TopicPublishInfo {
   
   
    public MessageQueue selectOneMessageQueue() {
   
   
        int index = this.sendWhichQueue.incrementAndGet();
        int pos = index % this.messageQueueList.size();
        return this.messageQueueList.get(pos);
    }
}

2 Consumer下的负载均衡

由于Consumer在广播模式下会把消息发给所有消费者,所以这里我们讨论的是集群模式下queue与消费者的负载均衡;

Rocketmq中规定一个Queue只能被一个消费者订阅,而一个消费者可以订阅多个Queue,这样自然会涉及到消费者与Queue分配协调策略。
当Broker扩容或缩容、Queue扩容等场景都可能导致消费者所订阅Topic的队列数量发生变化,也会再次重新分配。

2.1 分配策略算法

Rocketmq提供了多种Queue分配策略算法,如下

  • AllocateMessageQueueAveragely

    平均算法: 算出平均值,将连续的队列按平均值分配给每个消费者。
    如果能够整除,则按顺序将平均值个Queue分配,如果不能整除,则将多余出的Queue按照Consumer顺序逐个分配

20231226-02.png

  • AllocateMessageQueueAveragelyByCircle

    环形平均算法:将消费者按顺序形成一个环形,然后按照这个环形顺序逐个给消费者分配一个Queue

20231226-03.png

  • AllocateMessageQueueConsistentHash

    一致性hash算法:先将消费端的hash值放于环上,同时计算队列的hash值,以顺时针方向,分配给离队列hash值最近的一个消费者节点

20231226-04.png

2.2 源码分析

  • 1 创建DefaultMQPushConsumer时默认使用AllocateMessageQueueAveragely分配策略,即平均分配算法
  • 2 consumer调用start启动时会调用到RebalanceService.start()开启一个任务
  • 3 RebalanceService会调用到RebalanceImpl.rebalanceByTopic执行具体的平衡策略
  • 4 进而拿到所有的Consumer和Queue使用设置的AllocateMessageQueueStrategy算法去分配订阅关系

2.2.1 核心代码

/*
   1.初始化:默认使用AllocateMessageQueueAveragely算法分配Queue
 */
public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsumer {
   
   
    public DefaultMQPushConsumer(final String consumerGroup) {
   
   
        this(null, consumerGroup, null, new AllocateMessageQueueAveragely());
    }
}
/*
   2.开启一个RebalanceService任务执行分配策略
 */
public class MQClientInstance {
   
   
    public void start() throws MQClientException {
   
   
        synchronized (this) {
   
   
            switch (this.serviceState) {
   
   
                case CREATE_JUST:
                    ...
                    // Start rebalance service
                    this.rebalanceService.start();
                    ...
                default:
                    break;
            }
        }
    }
}
/*
   3.RebalanceImpl.rebalanceByTopic执行具体的分配逻辑
 */
public abstract class RebalanceImpl {
   
   
    private boolean rebalanceByTopic(final String topic, final boolean isOrder) {
   
   
        boolean balanced = true;
        switch (messageModel) {
   
   
            ...
            case CLUSTERING: {
   
   
                // 拿到所有的Queue
                Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
                // 拿到所有的消费者ID
                List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
                ...
                if (mqSet != null && cidAll != null) {
   
   
                    List<MessageQueue> mqAll = new ArrayList<>();
                    mqAll.addAll(mqSet);

                    Collections.sort(mqAll);
                    Collections.sort(cidAll);

                    // 初始化设置的分配算法:即AllocateMessageQueueAveragely平均分配算法
                    AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;

                    List<MessageQueue> allocateResult = null;
                    try {
   
   
                        // 调用分配算法的具体实现
                        allocateResult = strategy.allocate(...mqAll, cidAll);
                    } catch (Throwable e) {
   
   
                        log.error("allocate message queue exception. strategy name: {}, ex: {}", strategy.getName(), e);
                        return false;
                    }
                    ...
                }
                break;
            }
            default:
                break;
        }
        return balanced;
    }
}
/*
   4.执行AllocateMessageQueueAveragely平均分配算法的具体实现
 */
public class AllocateMessageQueueAveragely extends AbstractAllocateMessageQueueStrategy {
   
   
    @Override
    public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll, List<String> cidAll) {
   
   

        List<MessageQueue> result = new ArrayList<>();
        if (!check(consumerGroup, currentCID, mqAll, cidAll)) {
   
   
            return result;
        }

        int index = cidAll.indexOf(currentCID);
        int mod = mqAll.size() % cidAll.size();
        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;
    }
}

最后

至此,通过源码探索我们就大致了解了Producer及Consumer的负载均衡策略,其中Producer主要是通过轮询方式依次把消息发送到Queue,而Consumer默认使用的
平均分配算法去解决消费者节点与Queue之间的分配订阅关系。

Tip以下是本人经过多年的工作经验集成的JavaWeb脚手架,封装了各种通用的starter可开箱即用,同时列举了互联网各种高性能场景的使用示例。

// Git代码
https://gitee.com/yeeevip/yeee-memo
https://github.com/yeeevip/yeee-memo
相关实践学习
RocketMQ一站式入门使用
从源码编译、部署broker、部署namesrv,使用java客户端首发消息等一站式入门RocketMQ。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
5天前
|
消息中间件 数据可视化 Go
Rabbitmq 搭建使用案例 [附源码]
Rabbitmq 搭建使用案例 [附源码]
16 0
|
23天前
|
负载均衡
【SpringCloud】Ribbon负载均衡原理、负载均衡策略、饥饿加载
【SpringCloud】Ribbon负载均衡原理、负载均衡策略、饥饿加载
22 0
|
24天前
|
负载均衡 关系型数据库 分布式数据库
【PolarDB开源】PolarDB读写分离实践:优化读取性能与负载均衡策略
【5月更文挑战第26天】PolarDB是云原生关系型数据库,通过读写分离优化性能和扩展性。它设置主节点处理写操作,从节点处理读操作,异步复制保证数据一致性。优化读取性能的策略包括增加从节点数量、使用只读实例和智能分配读请求。负载均衡策略涉及基于权重、连接数和地理位置的分配。实践示例中,电商网站通过主从架构、只读实例和负载均衡策略提升商品查询效率。PolarDB的读写分离与负载均衡为企业应对大数据和高并发提供了有效解决方案。
136 0
|
27天前
|
域名解析 缓存 运维
【域名解析DNS专栏】DNS解析策略:如何实现负载均衡与故障转移
【5月更文挑战第23天】DNS在互联网中扮演关键角色,将域名转换为IP地址。本文探讨DNS的负载均衡和故障转移技术,以增强服务可用性和性能。负载均衡包括轮询(简单分配流量)和加权轮询(按服务器处理能力分配)。故障转移通过主备策略和TTL值实现快速切换,确保服务连续性。实践案例展示了在电商网站如何应用这些策略。DNS策略优化可提升网站速度和稳定性,借助云服务和智能工具,DNS管理更加高效。
【域名解析DNS专栏】DNS解析策略:如何实现负载均衡与故障转移
|
27天前
|
消息中间件 小程序 网络性能优化
蓝易云 - 直播小程序源码有用的协议知识:MQTT协
在直播小程序源码中,MQTT协议可以用于实现实时消息推送,如弹幕、聊天消息、礼物信息等。通过使用MQTT协议,可以确保消息的实时性和可靠性,从而提高用户体验。
55 0
|
1月前
|
消息中间件 存储 运维
深入理解MQ消息队列的高可用与可靠性策略
深入理解MQ消息队列的高可用与可靠性策略
1222 3
|
1月前
|
负载均衡
Ribbon负载均衡策略
Ribbon负载均衡策略
|
1月前
|
负载均衡 算法 Java
Ribbon的负载均衡策略
Ribbon的负载均衡策略
42 2
|
1月前
|
缓存 负载均衡 应用服务中间件
nginx的各种负载均衡策略与各种负载均衡策略如何配置
Nginx支持多种负载均衡策略,如轮询、加权轮询、IP哈希、最少连接、URL哈希和fair策略。轮询是默认策略,每个请求按顺序分发;加权轮询根据权重分配请求;IP哈希确保相同IP的请求始终发送到同一服务器;最少连接将请求发送给连接数最少的服务器;URL哈希(需额外工具或模块)和fair策略则依据URL和响应时间分配请求。配置变更需更新nginx.conf并重新加载或重启服务,具体配置应参照官方文档。
67 0
|
1月前
|
消息中间件 Java RocketMQ
【深度挖掘 RocketMQ底层源码】「底层源码挖掘系列」抽丝剥茧贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-下)
【深度挖掘 RocketMQ底层源码】「底层源码挖掘系列」抽丝剥茧贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-下)
16 1

热门文章

最新文章