RocketMQ 消费自动负载均衡机制

1. 背景介绍

RocketMQ是一款具备低延时、高性能的消息队列产品。通过namesrv这个功能简单的组件，将松散的不同broker连起来，组成一个分布式的消息队列集群。本文是基于4.9.4版本的RocketMQ进行分析。

1.1. 消费模式

RocketMQ有两种消费模式:集群消费和广播消费

集群消费模式: 一个消息只被订阅组内的一个消费者实例消费。不同消费者实例互相分摊消息，实现消费负载均衡

广播消费模式: 一个消费会被订阅组内的所有消费者实例消费。

要实现消费的自动负载均衡，只能使用集群消费模式，也是强烈推荐使用的消费模式。

1.2 队列(queue)

消息都存放在主题(topic)上。而一个主题可以有多个队列(queue)。同一个主题，可以分布在不同的broker上。
所以主题的队列数，是主题在不同broker上的队列数总和。
broker往namesrv上注册所有主题的信息。这样客户端通过namesrv可以看到主题的全貌。
而队列的主要作用是解决客户端的并发消费。

2. 消费负载均衡的功能介绍

单个消费进程是有性能的上限的。不同的业务场景，消耗的时间不同。短则毫秒级，长则可能几秒甚至几分钟。为了提高消费的性能，需要可以并行去消费消息，而且并行消费期间，消费者实例的消费进度和运行情况应该不能互相干扰。

为了解决这个问题，需要实现消费端的自动负载均衡。上一节有提到，队列正是解决这个问题的关键。

下面的JSON是订阅组sub_1保存topic_1主题的消费进度的文件保存信息。可见，消费进度，其实正是具体每个队列的消息位置(offset)

rocketmq内部使用JSON格式保存所有的消费进度，一般是放在数据存储目录下config/consumerOffset.json 文件里面

"topic_1@sub_1":{0:181,1:178,2:176,3:183,4:184,5:182,6:177,7:179}

为了不互相干扰也可以并发去消费，客户端需要分配queue的订阅,让每个客户端都去订阅不同的queue，没有重叠，也没有漏了。达到如下的效果：

客户端的数量比队列数多的时候。超出的客户端会无法订阅、消费任何消息。直到正在订阅的客户端停止。

3. 消费负载均衡的原理分析

结合客户端的源代码进行。

其中核心代码如下：

1. RebalanceService.java：客户端自带的服务，默认每20秒进行一次doRebalance();
2. RebalanceImpl.java: 针对每个订阅组的每个主题，进行重新负载均衡
3. MQClientInstance.java: 客户端对象，默认每30秒从namesrv更新主题的路由信息。
4. AllocateMessageQueueAveragely.java: 负载均衡策略

核心的算法如下：
1. 客户端启动的时候，会从namesrv获取订阅主题的路由信息。后续默认每30秒更新一次
2. 客户端默认每20秒进行重新负载均衡计算。
2.1. 消费的负载均衡计算，本质是将每个订阅组里面的每个主题都进行负载计算
2.2. 单个订阅组的主题的负载计算方法如下：基于步骤1得到主题的所有队列信息，以及从服务端获取所有客户端ID列表。将两者排序，确保每个客户端都得到相同的结果。再基于负载均衡策略进行计算。
2.3 负载均衡的策略：默认使用 AllocateMessageQueueAveragely策略，也即是上面例子的介绍按顺序平均分配队列的方式。
* 可以根据需要设置不同的策略，比如一致性HASH等。但要确保每个客户端都要有相同的配置。

下面的相关的代码。
RebalanceService.java：

public class RebalanceService extends ServiceThread {
    private static long waitInterval =
        Long.parseLong(System.getProperty(
            "rocketmq.client.rebalance.waitInterval", "20000"));
    private final InternalLogger log = ClientLogger.getLog();
    private final MQClientInstance mqClientFactory;

    public RebalanceService(MQClientInstance mqClientFactory) {
        this.mqClientFactory = mqClientFactory;
    }

    @Override
    public void run() {
        log.info(this.getServiceName() + " service started");

        while (!this.isStopped()) {
            this.waitForRunning(waitInterval);
            this.mqClientFactory.doRebalance();
        }

        log.info(this.getServiceName() + " service end");
    }

    @Override
    public String getServiceName() {
        return RebalanceService.class.getSimpleName();
    }
}

RebalanceImpl.doRebalance(): 对所有订阅组订阅的所有TOPIC进行重新负载计算
RebalanceImpl.rebalanceByTopic()：对具体每个topic进行重新负载计算

public void doRebalance(final boolean isOrder) {
    Map<String, SubscriptionData> subTable = this.getSubscriptionInner();
    if (subTable != null) {
        for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {
            final String topic = entry.getKey();
            try {
                this.rebalanceByTopic(topic, isOrder);
            } catch (Throwable e) {
                if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                    log.warn("rebalanceByTopic Exception", e);
                }
            }
        }
    }

    this.truncateMessageQueueNotMyTopic();
}

private void rebalanceByTopic(final String topic, final boolean isOrder) {
    switch (messageModel) {
        case BROADCASTING: {
            Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
            if (mqSet != null) {
                boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder);
                if (changed) {
                    this.messageQueueChanged(topic, mqSet, mqSet);
                    log.info("messageQueueChanged {} {} {} {}",
                        consumerGroup,
                        topic,
                        mqSet,
                        mqSet);
                }
            } else {
                log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);
            }
            break;
        }
        case CLUSTERING: {
            Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
            List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
            if (null == mqSet) {
                if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                    log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);
                }
            }

            if (null == cidAll) {
                log.warn("doRebalance, {} {}, get consumer id list failed", consumerGroup, topic);
            }

            if (mqSet != null && cidAll != null) {
                List<MessageQueue> mqAll = new ArrayList<MessageQueue>();
                mqAll.addAll(mqSet);

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

                AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;

                List<MessageQueue> allocateResult = null;
                try {
                    allocateResult = strategy.allocate(
                        this.consumerGroup,
                        this.mQClientFactory.getClientId(),
                        mqAll,
                        cidAll);
                } catch (Throwable e) {
                    log.error("AllocateMessageQueueStrategy.allocate Exception. allocateMessageQueueStrategyName={}", strategy.getName(),
                        e);
                    return;
                }

                Set<MessageQueue> allocateResultSet = new HashSet<MessageQueue>();
                if (allocateResult != null) {
                    allocateResultSet.addAll(allocateResult);
                }

                boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet, isOrder);
                if (changed) {
                    log.info(
                        "rebalanced result changed. allocateMessageQueueStrategyName={}, group={}, topic={}, clientId={}, mqAllSize={}, cidAllSize={}, rebalanceResultSize={}, rebalanceResultSet={}",
                        strategy.getName(), consumerGroup, topic, this.mQClientFactory.getClientId(), mqSet.size(), cidAll.size(),
                        allocateResultSet.size(), allocateResultSet);
                    this.messageQueueChanged(topic, mqSet, allocateResultSet);
                }
            }
            break;
        }
        default:
            break;
    }
}

MQClientInstance.startScheduledTask():启动一系列的定时任务。其中包括默认每30秒从namesrv获取最新的主题路由信息

private void startScheduledTask() {
    if (null == this.clientConfig.getNamesrvAddr()) {
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                try {
                    MQClientInstance.this.mQClientAPIImpl.fetchNameServerAddr();
                } catch (Exception e) {
                    log.error("ScheduledTask fetchNameServerAddr exception", e);
                }
            }
        }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
    }

    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            try {
                MQClientInstance.this.updateTopicRouteInfoFromNameServer();
            } catch (Exception e) {
                log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e);
            }
        }
    }, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);

    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            try {
                MQClientInstance.this.cleanOfflineBroker();
                MQClientInstance.this.sendHeartbeatToAllBrokerWithLock();
            } catch (Exception e) {
                log.error("ScheduledTask sendHeartbeatToAllBroker exception", e);
            }
        }
    }, 1000, this.clientConfig.getHeartbeatBrokerInterval(), TimeUnit.MILLISECONDS);

    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            try {
                MQClientInstance.this.persistAllConsumerOffset();
            } catch (Exception e) {
                log.error("ScheduledTask persistAllConsumerOffset exception", e);
            }
        }
    }, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);

    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            try {
                MQClientInstance.this.adjustThreadPool();
            } catch (Exception e) {
                log.error("ScheduledTask adjustThreadPool exception", e);
            }
        }
    }, 1, 1, TimeUnit.MINUTES);
}

AllocateMessageQueueAveragely.allocate(): 分配消费负载

@Override
public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,
    List<String> cidAll) {

    List<MessageQueue> result = new ArrayList<MessageQueue>();
    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;
}

4. 默认消费的负载均衡的机制分析和问题

分布式消息RocketMQ采取了一种比较松散的架构。通过namesrv把多个独立运行broker信息汇总在一起。而客户端进行负载均衡的时候，也是独立计算和运行的。服务端实际没有参与其中。
优点是充分利用了分布式的特征，可以非常线性的增加性能，可以支撑一个非常大规模的集群和客户端数量也没问题。
缺点也有如下几个：
a. 负载均衡的计算并不是同时进行,也没有统一进行分配，都是客户端自行独立计算。因而很容易出现的队列分配重叠情况。
b. 不能马上感知到集群的消费者实例或主题的变化，存在一定的延时和滞后。实际负载是经过一定时间最终达到负载均衡分配而已。
c. 消费者的消费进度是客户端定时上传更新的。对新加入的消费者实例而言，自行计算到分配的队列后，从broker上获取的消费进度，可能也有一定的延时。
d. 消费者实例比队列数多的情况，肯定存在消费者实例无法订阅消费任何消息的情况。为了缓解这个问题，可以一开始创建更多的队列数。但队列数多了也会占用更多的资源，对性能会有一定的影响。
f. 存在单个队列消息量过多，消费无法负载分摊的问题。

以上的问题a,c都会导致消费重复的问题。因而要正确的使用RocketMQ，需要确保业务侧实现消息的幂等性处理。

5. 展望未来

既然发现了这么多问题，那是否有一个从根本上解决上述问题的方法或思路呢。答案在RocketMQ 5.0 引入的新消费接口上，把消费负载分配的逻辑都放到了broker上。有兴趣的可以自行查看RIP 19的描述。

应用商城

合作伙伴

开发者

支持与服务

了解天翼云

RocketMQ消费负载均衡

RocketMQ 消费自动负载均衡机制

1. 背景介绍

1.1. 消费模式

1.2 队列(queue)

2. 消费负载均衡的功能介绍

3. 消费负载均衡的原理分析

4. 默认消费的负载均衡的机制分析和问题

5. 展望未来

活动

应用商城

合作伙伴

开发者

支持与服务

了解天翼云

RocketMQ消费负载均衡

RocketMQ 消费自动负载均衡机制

1. 背景介绍

1.1. 消费模式

1.2 队列(queue)

2. 消费负载均衡的功能介绍

3. 消费负载均衡的原理分析

4. 默认消费的负载均衡的机制分析和问题

5. 展望未来