【RocketMQ每日一问】RocketMQ重平衡逻辑是怎么样的?

2024-01-10 06:11:25
  1. 消费者启动时,订阅相应的topic并加入到消费者组。
  2. 消费者将消费进度信息存储到Broker中,包括当前消费到的消息的offset、队列信息等。
  3. 消费者定时从Broker中获取topic的路由信息(包括消息队列、broker信息等),并更新本地缓存。
  4. 当消费者组内新增或删除消费者时,触发重平衡事件,即重新分配消息队列给消费者。
  5. 重平衡事件由消费组内最先触发的消费者发起,通知Broker开始重平衡。Broker接到重平衡请求后向消费者组内的其他消费者发送通知。
  6. 消费者接收到重平衡通知后开始重新计算自己所分配到的队列,根据拉取到的消息队列的分配策略重新分配队列。
  7. 消费者重新分配完队列后将分配结果发送给Broker,并更新本地缓存的路由信息。
  8. Broker接收到所有消费者发送的分配结果后,按照分配结果更新自己记录的消费组的消费进度信息,包括消费进度、offset等信息。
  9. 当消费者消费完消息队列中的所有消息后,将消费进度信息更新到Broker中,以便下次重平衡时可以根据消费进度信息进行分配。

以下是重平衡代码:

private boolean rebalanceByTopic(final String topic, final boolean isOrder) {
        boolean balanced = true;
        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);
                    }

                    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;
            }
            case CLUSTERING: {
                //内存中拿出所有queue
                Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
                //GET_CONSUMER_LIST_BY_GROUP这个请求去ns拿出所有的客户端
                List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
                if (null == mqSet) {
                    if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                        this.messageQueueChanged(topic, Collections.<MessageQueue>emptySet(), Collections.<MessageQueue>emptySet());
                        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);
                    //将两个list进行排序
                    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("allocate message queue exception. strategy name: {}, ex: {}", strategy.getName(), e);
                        return false;
                    }

                    Set<MessageQueue> allocateResultSet = new HashSet<MessageQueue>();
                    if (allocateResult != null) {
                        allocateResultSet.addAll(allocateResult);
                    }
                    //对比queue是否有变化,并且更新到processQueueTable中
                    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));
                }
                break;
            }
            default:
                break;
        }

        return balanced;
    }

文章来源:https://blog.csdn.net/jianjun_fei/article/details/135486538
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。