RocketMQ消费者如何实现重平衡

this.rebalanceService.start();
复制代码

public void start() {
        log.info("Try to start service thread:{} started:{} lastThread:{}", getServiceName(), started.get(), thread);
        if (!started.compareAndSet(false, true)) {
            return;
        }
        stopped = false;
        // 创建线程
        this.thread = new Thread(this, getServiceName());
        this.thread.setDaemon(isDaemon);
        // 启动线程
        this.thread.start();
    }
复制代码

this.rebalanceService = new RebalanceService(this);
复制代码

// 默认20秒，可以通过rocketmq.client.rebalance.waitInterval配置调整
    private static long waitInterval =
        Long.parseLong(System.getProperty(
            "rocketmq.client.rebalance.waitInterval", "20000"));  
    @Override
    public void run() {
        log.info(this.getServiceName() + " service started");
        while (!this.isStopped()) {
            // 等待20秒
            this.waitForRunning(waitInterval);
            // 执行重平衡
            this.mqClientFactory.doRebalance();
        }
        log.info(this.getServiceName() + " service end");
    }
复制代码

public boolean doRebalance(final boolean isOrder) {
        boolean balanced = true;
        Map<String, SubscriptionData> subTable = this.getSubscriptionInner();
        if (subTable != null) {
            for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {
                final String topic = entry.getKey();
                try {
                    // 1.先判断是否是客户端重平衡
                    // 2.如果不是客户端重平衡，就要判断这个topic是否存在broker里面，因为要使用boker重平衡，不在这篇文章讨论范围内
                    if (!clientRebalance(topic) && tryQueryAssignment(topic)) {
                        balanced = this.getRebalanceResultFromBroker(topic, isOrder);
                    } else {
                        // 客户端重平衡实现
                        balanced = this.rebalanceByTopic(topic, isOrder);
                    }
                } catch (Throwable e) {
                    if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                        log.warn("rebalance Exception", e);
                        balanced = false;
                    }
                }
            }
        }
        // 更新重平衡后的processQueueTable、popProcessQueueTable、topicClientRebalance、topicBrokerRebalance
        this.truncateMessageQueueNotMyTopic();
        return balanced;
    }
复制代码

case BROADCASTING: {
    // 根据topic获取所有的MessageQueue实例
    Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
    if (mqSet != null) {
        // 如果MessageQueue列表有变化，那么就更新processQueueTable
        // 剔除下线的消费者与超时的broker
        // 给新上线的broker创建pullRequest准备拉消息
        boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder);
        if (changed) {
            // 调用监听器的messageQueueChanged()方法
            this.messageQueueChanged(topic, mqSet, mqSet);
            log.info("messageQueueChanged {} {} {} {}", consumerGroup, topic, mqSet, mqSet);
        }
        // 对比当前正在运行的MessageQueue，如果和mqSet一致，说明已经达到平衡了
        balanced = mqSet.equals(getWorkingMessageQueue(topic));
    } else {
        this.messageQueueChanged(topic, Collections.<MessageQueue>emptySet(), Collections.<MessageQueue>emptySet());
        log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);
    }
    break;
}
复制代码

this.scheduledExecutorService.scheduleAtFixedRate(() -> {
            try {
                MQClientInstance.this.updateTopicRouteInfoFromNameServer();
            } catch (Exception e) {
                log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e);
            }
        }, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);
复制代码

// topic对应的所有MessageQueue
Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
// 同一个消费组下订阅了topic的消费者ConsumerId列表
List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
复制代码

List<MessageQueue> mqAll = new ArrayList<MessageQueue>();
mqAll.addAll(mqSet);
// 排序MessageQueue
Collections.sort(mqAll);
// 排序ConsumerId
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("allocate message queue exception. strategy name: {}, ex: {}", strategy.getName(), e);
    return false;
}
复制代码

@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;
        }
        // 获取当前消费者consumerId所在的索引位
        int index = cidAll.indexOf(currentCID);
        int mod = mqAll.size() % cidAll.size();
        // 计算每个消费者平均能分配到多少个MessageQueue
        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;
    }
复制代码

boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet, isOrder);
if (changed) {
    log.info(
                            "client 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);
}
balanced = allocateResultSet.equals(getWorkingMessageQueue(topic));
复制代码