RocketMQ架构08：消息路由与负载均衡 - 智能路由的奥秘

public class TopicRouteData {
    // Queue 数据列表
    private List<QueueData> queueDatas;

    // Broker 数据列表
    private List<BrokerData> brokerDatas;

    // 过滤服务器表
    private HashMap<String, List<String>> filterServerTable;
}

// Queue 数据
public class QueueData {
    private String brokerName;       // Broker 名称
    private int readQueueNums;       // 读队列数
    private int writeQueueNums;      // 写队列数
    private int perm;                // 权限
    private int topicSynFlag;        // 同步标志
}

// Broker 数据
public class BrokerData {
    private String cluster;          // 集群名
    private String brokerName;       // Broker 名称
    private HashMap<Long, String> brokerAddrs;  // Broker 地址表
    // Key: brokerId (0=Master, >0=Slave)
    // Value: broker 地址
}

{
  "queueDatas": [
    {
      "brokerName": "broker-a",
      "readQueueNums": 4,
      "writeQueueNums": 4,
      "perm": 6
    },
    {
      "brokerName": "broker-b",
      "readQueueNums": 4,
      "writeQueueNums": 4,
      "perm": 6
    }
  ],
  "brokerDatas": [
    {
      "cluster": "DefaultCluster",
      "brokerName": "broker-a",
      "brokerAddrs": {
        "0": "192.168.1.10:10911",  // Master
        "1": "192.168.1.11:10911"   // Slave
      }
    },
    {
      "cluster": "DefaultCluster",
      "brokerName": "broker-b",
      "brokerAddrs": {
        "0": "192.168.1.20:10911",
        "1": "192.168.1.21:10911"
      }
    }
  ]
}

// Producer 路由更新
public class MQClientInstance {

    // 更新单个 Topic 的路由信息
    public boolean updateTopicRouteInfoFromNameServer(final String topic) {
        try {
            // 1. 从 NameServer 获取路由信息
            TopicRouteData topicRouteData =
                this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3);

            if (topicRouteData != null) {
                // 2. 对比路由是否变化
                TopicRouteData old = this.topicRouteTable.get(topic);
                boolean changed = topicRouteDataIsChange(old, topicRouteData);

                if (!changed) {
                    changed = this.isNeedUpdateTopicRouteInfo(topic);
                }

                if (changed) {
                    // 3. 更新 Producer 的路由表
                    for (Entry<String, MQProducerInner> entry : this.producerTable.entrySet()) {
                        MQProducerInner impl = entry.getValue();
                        if (impl != null) {
                            impl.updateTopicPublishInfo(topic, publishInfo);
                        }
                    }

                    // 4. 更新 Consumer 的路由表
                    for (Entry<String, MQConsumerInner> entry : this.consumerTable.entrySet()) {
                        MQConsumerInner impl = entry.getValue();
                        if (impl != null) {
                            impl.updateTopicSubscribeInfo(topic, subscribeInfo);
                        }
                    }

                    // 5. 更新本地缓存
                    this.topicRouteTable.put(topic, cloneTopicRouteData(topicRouteData));

                    log.info("topicRouteTable.put TopicRouteData[{}]", topic);
                }

                return true;
            }
        } catch (Exception e) {
            log.error("updateTopicRouteInfoFromNameServer Exception", e);
        }

        return false;
    }

    // 定时更新所有 Topic 的路由
    public void startScheduledTask() {
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    MQClientInstance.this.updateTopicRouteInfoFromNameServer();
                } catch (Exception e) {
                    log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e);
                }
            }
        }, 10, 30, TimeUnit.SECONDS);  // 每30秒更新一次
    }
}

public class TopicPublishInfo {
    // Queue 列表
    private List<MessageQueue> messageQueueList = new ArrayList<>();

    // 上次选择的 Queue 索引
    private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex();

    // 轮询选择 Queue
    public MessageQueue selectOneMessageQueue() {
        int index = this.sendWhichQueue.incrementAndGet();
        int pos = Math.abs(index) % this.messageQueueList.size();
        if (pos < 0) {
            pos = 0;
        }
        return this.messageQueueList.get(pos);
    }

    // 故障规避选择 Queue
    public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
        if (lastBrokerName == null) {
            return selectOneMessageQueue();
        } else {
            // 循环查找，跳过上次失败的 Broker
            for (int i = 0; i < this.messageQueueList.size(); i++) {
                int index = this.sendWhichQueue.incrementAndGet();
                int pos = Math.abs(index) % this.messageQueueList.size();
                if (pos < 0) {
                    pos = 0;
                }
                MessageQueue mq = this.messageQueueList.get(pos);

                // 跳过故障 Broker
                if (!mq.getBrokerName().equals(lastBrokerName)) {
                    return mq;
                }
            }
            // 如果所有 Broker 都故障，返回任意一个
            return selectOneMessageQueue();
        }
    }
}

public class MQFaultStrategy {

    // Broker 故障延迟表
    private final ConcurrentHashMap<String, FaultItem> faultItemTable =
        new ConcurrentHashMap<>(16);

    // Broker 是否可用
    public boolean isAvailable(final String brokerName) {
        final FaultItem faultItem = this.faultItemTable.get(brokerName);
        if (faultItem != null) {
            return faultItem.isAvailable();
        }
        return true;
    }

    // 更新 Broker 故障信息
    public void updateFaultItem(final String brokerName, final long currentLatency,
                                boolean isolation) {
        FaultItem old = this.faultItemTable.get(brokerName);
        if (null == old) {
            final FaultItem faultItem = new FaultItem(brokerName);
            faultItem.setCurrentLatency(currentLatency);
            faultItem.setStartTimestamp(System.currentTimeMillis());
            old = this.faultItemTable.putIfAbsent(brokerName, faultItem);
            if (old != null) {
                old.setCurrentLatency(currentLatency);
                old.setStartTimestamp(System.currentTimeMillis());
            }
        } else {
            old.setCurrentLatency(currentLatency);
            old.setStartTimestamp(System.currentTimeMillis());
        }
    }

    // 选择可用的 MessageQueue
    public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo,
                                             final String lastBrokerName) {
        // 1. 启用故障延迟
        if (this.sendLatencyFaultEnable) {
            try {
                // 轮询选择
                int index = tpInfo.getSendWhichQueue().incrementAndGet();
                for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
                    int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
                    MessageQueue mq = tpInfo.getMessageQueueList().get(pos);

                    // 检查 Broker 是否可用
                    if (isAvailable(mq.getBrokerName())) {
                        return mq;
                    }
                }

                // 如果没有可用的，选择延迟最小的
                final String notBestBroker = pickOneAtLeast();
                int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
                if (writeQueueNums > 0) {
                    final MessageQueue mq = tpInfo.selectOneMessageQueue();
                    if (notBestBroker != null) {
                        mq.setBrokerName(notBestBroker);
                        mq.setQueueId(tpInfo.getSendWhichQueue().incrementAndGet() % writeQueueNums);
                    }
                    return mq;
                } else {
                    this.faultItemTable.remove(notBestBroker);
                }
            } catch (Exception e) {
                log.error("Error occurred when selecting message queue", e);
            }

            return tpInfo.selectOneMessageQueue();
        }

        // 2. 不启用故障延迟（默认）
        return tpInfo.selectOneMessageQueue(lastBrokerName);
    }
}

public class DefaultMQProducerImpl {

    // 发送消息（带重试）
    private SendResult sendDefaultImpl(
        Message msg,
        final CommunicationMode communicationMode,
        final SendCallback sendCallback,
        final long timeout
    ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {

        // 1. 获取路由信息
        TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());

        if (topicPublishInfo != null && topicPublishInfo.ok()) {
            boolean callTimeout = false;
            MessageQueue mq = null;
            Exception exception = null;
            SendResult sendResult = null;

            // 2. 重试次数
            int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;

            int times = 0;
            String[] brokersSent = new String[timesTotal];

            // 3. 循环重试
            for (; times < timesTotal; times++) {
                String lastBrokerName = null == mq ? null : mq.getBrokerName();

                // 选择 MessageQueue（故障规避）
                MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
                if (mqSelected != null) {
                    mq = mqSelected;
                    brokersSent[times] = mq.getBrokerName();

                    try {
                        beginTimestampPrev = System.currentTimeMillis();

                        // 4. 发送消息
                        sendResult = this.sendKernelImpl(
                            msg,
                            mq,
                            communicationMode,
                            sendCallback,
                            topicPublishInfo,
                            timeout - costTime
                        );

                        endTimestamp = System.currentTimeMillis();

                        // 5. 更新故障延迟
                        this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);

                        // 发送成功，直接返回
                        return sendResult;

                    } catch (RemotingException e) {
                        endTimestamp = System.currentTimeMillis();
                        this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
                        exception = e;
                        continue;
                    } catch (MQBrokerException e) {
                        endTimestamp = System.currentTimeMillis();
                        this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
                        exception = e;

                        // 某些异常不重试
                        if (!retryAnotherBrokerWhenNotStoreOK) {
                            throw e;
                        }
                        continue;
                    }
                } else {
                    break;
                }
            }

            // 6. 重试耗尽，抛出异常
            throw new MQClientException("Send failed", exception);
        }

        throw new MQClientException("No route info of this topic: " + msg.getTopic());
    }
}

public interface AllocateMessageQueueStrategy {
    /**
     * 分配队列
     * @param consumerGroup 消费组
     * @param currentCID 当前消费者ID
     * @param mqAll 所有队列
     * @param cidAll 所有消费者ID列表
     * @return 分配给当前消费者的队列列表
     */
    List<MessageQueue> allocate(
        final String consumerGroup,
        final String currentCID,
        final List<MessageQueue> mqAll,
        final List<String> cidAll
    );
}

public class AllocateMessageQueueAveragely implements AllocateMessageQueueStrategy {

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

        List<MessageQueue> result = new ArrayList<>();

        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;
    }
}

public class AllocateMessageQueueAveragelyByCircle implements AllocateMessageQueueStrategy {

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

        List<MessageQueue> result = new ArrayList<>();

        int index = cidAll.indexOf(currentCID);

        // 环形分配
        for (int i = index; i < mqAll.size(); i++) {
            if (i % cidAll.size() == index) {
                result.add(mqAll.get(i));
            }
        }

        return result;
    }
}

// Consumer 下线触发 Rebalance
public class RebalanceService extends ServiceThread {

    @Override
    public void run() {
        while (!this.isStopped()) {
            this.waitForRunning(waitInterval);
            this.mqClientFactory.doRebalance();
        }
    }
}

// Rebalance 实现
public boolean doRebalance(final boolean isOrder) {
    boolean balanced = true;

    // 遍历所有订阅的 Topic
    Map<String, SubscriptionData> subTable = this.getSubscriptionInner();
    if (subTable != null) {
        for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {
            final String topic = entry.getKey();
            try {
                // 对每个 Topic 执行 Rebalance
                balanced = this.rebalanceByTopic(topic, isOrder);
            } catch (Throwable e) {
                log.error("rebalanceByTopic Exception", e);
            }
        }
    }

    this.truncateMessageQueueNotMyTopic();

    return balanced;
}

# Producer 配置优化
retryTimesWhenSendFailed=2          # 同步发送失败重试次数
retryTimesWhenSendAsyncFailed=2     # 异步发送失败重试次数
sendLatencyFaultEnable=true         # 启用延迟故障容错

# Consumer 配置优化
consumeThreadMin=20                 # 最小消费线程数
consumeThreadMax=64                 # 最大消费线程数
pullBatchSize=32                    # 批量拉取数量
allocateMessageQueueStrategy=       # 分配策略
  AllocateMessageQueueAveragely