private void rebalanceByTopic(final String topic, final boolean isOrder) {
    // 获取所有 MessageQueue
    Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
    // 获取当前 group 所有在线的消费者
    List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);

    ...

    // 排序
    Collections.sort(mqAll);
    Collections.sort(cidAll);

    // 获取当前客户端配置的负载均衡策略
    AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;

    List<MessageQueue> allocateResult = null;

    ...

    // 根据指定的负载均衡策略计算自己要负责的队列
    allocateResult = strategy.allocate(
        this.consumerGroup,
        this.mQClientFactory.getClientId(),
        mqAll,

    ...

    // 根据计算结果创建 ProcessQueue （用于拉取、消费消息的数据结构）
    boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet, isOrder);
    if (changed) {
        ...

        // 如果 ProcessQueue 有更新则进行一些处理工作
        this.messageQueueChanged(topic, mqSet, allocateResultSet);
    }
}

private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet, final boolean isOrder) {
    // 删除不再负责的 MessageQueue 对应的 ProcessQueue
    Iterator<Entry<MessageQueue, ProcessQueue>> it = this.processQueueTable.entrySet().iterator();
    while (it.hasNext()) {
        Entry<MessageQueue, ProcessQueue> next = it.next();
        MessageQueue mq = next.getKey();
        ProcessQueue pq = next.getValue();

        if (mq.getTopic().equals(topic)) {
            if (!mqSet.contains(mq)) {
                // 停止 ProcessQueue 的消费
                pq.setDropped(true);
                // 向 Broker 更新 offset，如果是顺序消费会释放申请的锁
                if (this.removeUnnecessaryMessageQueue(mq, pq)) {
                    it.remove();
                    changed = true;
                    log.info("doRebalance, {}, remove unnecessary mq, {}", consumerGroup, mq);
                }
            }
        }
    }

    // 为新分配的 MessageQueue 创建 ProcessQueue
    List<PullRequest> pullRequestList = new ArrayList<PullRequest>();
    for (MessageQueue mq : mqSet) {
        if (!this.processQueueTable.containsKey(mq)) {
            ...

            ProcessQueue pq = new ProcessQueue();

            ...

            if (nextOffset >= 0) {
                ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
            }
            ...
        }
    }
}

public boolean registerConsumer(final String group, final ClientChannelInfo clientChannelInfo,
    ConsumeType consumeType, MessageModel messageModel, ConsumeFromWhere consumeFromWhere,
    final Set<SubscriptionData> subList, boolean isNotifyConsumerIdsChangedEnable) {

    ...

    // 是否是新客户端
    boolean r1 = consumerGroupInfo.updateChannel(clientChannelInfo, consumeType, messageModel, consumeFromWhere);

    // 更新订阅数据，当订阅新 topic 或有旧的 topic 不再订阅时返回 true
    boolean r2 = consumerGroupInfo.updateSubscription(subList);

    if (r1 || r2) {
        // 可以在 BrokerConfig 中配置，默认为 true
        if (isNotifyConsumerIdsChangedEnable) {
            // 通知客户端进行负载均衡
            this.consumerIdsChangeListener.handle(ConsumerGroupEvent.CHANGE, group, consumerGroupInfo.getAllChannel());
        }
    }

    ...
}

@Override
public void run() {
    ...

    final ChannelEventListener listener = NettyRemotingAbstract.this.getChannelEventListener();

    while (!this.isStopped()) {
        try {
            NettyEvent event = this.eventQueue.poll(3000, TimeUnit.MILLISECONDS);
            if (event != null && listener != null) {
                switch (event.getType()) {
                    case IDLE:
                        listener.onChannelIdle(event.getRemoteAddr(), event.getChannel());
                        break;
                    case CLOSE:
                        listener.onChannelClose(event.getRemoteAddr(), event.getChannel());
                        break;
                    case CONNECT:
                        listener.onChannelConnect(event.getRemoteAddr(), event.getChannel());
                        break;
                    case EXCEPTION:
                        listener.onChannelException(event.getRemoteAddr(), event.getChannel());
                        break;
                    default:
                        break;

                }
            }
        } catch (Exception e) {
            log.warn(this.getServiceName() + " service has exception. ", e);
        }
    }
}

public void scanNotActiveChannel() {
Iterator<Entry<Channel, ClientChannelInfo>> itChannel = channelInfoTable.entrySet().iterator();
      ...

      while (itChannel.hasNext()) {
          Entry<Channel, ClientChannelInfo> nextChannel = itChannel.next();
          ClientChannelInfo clientChannelInfo = nextChannel.getValue();
          long diff = System.currentTimeMillis() - clientChannelInfo.getLastUpdateTimestamp();
          // 默认两分钟
          if (diff > CHANNEL_EXPIRED_TIMEOUT) {
              // 关闭连接
              RemotingUtil.closeChannel(clientChannelInfo.getChannel());
              itChannel.remove();
          }
      }

      ...
}