RocketMQ源码 — 六、 RocketMQ高可用（1）

// org.apache.rocketmq.broker.BrokerController#registerBrokerAll
public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway) {
    TopicConfigSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildTopicConfigSerializeWrapper();

    // 省略中间代码...
    RegisterBrokerResult registerBrokerResult = this.brokerOuterAPI.registerBrokerAll(
        this.brokerConfig.getBrokerClusterName(),
        this.getBrokerAddr(),
        this.brokerConfig.getBrokerName(),
        this.brokerConfig.getBrokerId(),
        this.getHAServerAddr(),
        topicConfigWrapper,
        this.filterServerManager.buildNewFilterServerList(),
        oneway,
        this.brokerConfig.getRegisterBrokerTimeoutMills());
    // 省略中间代码...
}

{
    "dataVersion": {
        "counter": 2,
        "timestamp": 1514252649572
    },
    "topicConfigTable": {
        "TopicTest": {
            "order": false,
            "perm": 6,
            "readQueueNums": 4,
            "topicFilterType": "SINGLE_TAG",
            "topicName": "TopicTest",
            "topicSysFlag": 0,
            "writeQueueNums": 4
        },
        "%RETRY%please_rename_unique_group_name_4": {
            "order": false,
            "perm": 6,
            "readQueueNums": 1,
            "topicFilterType": "SINGLE_TAG",
            "topicName": "%RETRY%please_rename_unique_group_name_4",
            "topicSysFlag": 0,
            "writeQueueNums": 1
        }
    }
}

// org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#registerBroker
public RegisterBrokerResult registerBroker(
    final String clusterName,
    final String brokerAddr,
    final String brokerName,
    final long brokerId,
    final String haServerAddr,
    final TopicConfigSerializeWrapper topicConfigWrapper,
    final List<String> filterServerList,
    final Channel channel) {
    RegisterBrokerResult result = new RegisterBrokerResult();
    try {
        try {
            // 省略中间代码...
            // 这里会判断只有master才会创建QueueData，因为只有master才包含了读写队列的信息
            // slave没有自己独立的读写队列信息（salve不会创建自己的queue信息），只是和master的的读写队列信息一致
            if (null != topicConfigWrapper
                && MixAll.MASTER_ID == brokerId) {
                if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())
                    || registerFirst) {
                    ConcurrentMap<String, TopicConfig> tcTable =
                        topicConfigWrapper.getTopicConfigTable();
                    if (tcTable != null) {
                        for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) {
                            // 这个方法创建了QueueData，QueueData包含broker set下的读写队列的信息
                            this.createAndUpdateQueueData(brokerName, entry.getValue());
                        }
                    }
                }
            }

            // 省略中间代码...
    } catch (Exception e) {
        log.error("registerBroker Exception", e);
    }

    return result;
}

// 每个cluster下的broker set信息，一个brokerName对应的broker set
private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
// 每个broker set中的broker信息(set中有哪些broker，每个broker的brokerId和brokerAddr)
private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
// 每个broker的存活情况
private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
// 每个topic下的queue信息，包括每个broker set中读写队列的个数，consumer消费消息和producer发送消息的路由信息都从这个数据结构中获取
private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;

// org.apache.rocketmq.client.impl.factory.MQClientInstance#updateTopicRouteInfoFromNameServer(java.lang.String, boolean, org.apache.rocketmq.client.producer.DefaultMQProducer)
public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,
    DefaultMQProducer defaultMQProducer) {
    try {
        if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
            try {
                TopicRouteData topicRouteData;
                if (isDefault && defaultMQProducer != null) {
                    // 省略中间代码...
                } else {
                    // 从manesrv获取topic的路由信息，namesrv从topicQueueTable获取到该topic对应的所有的QueueData
                    // 然后将每个brokerName下的BrokerData返回
                    topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3);
                }
                      // 省略中间代码...
                      for (BrokerData bd : topicRouteData.getBrokerDatas()) {
                            // 每个broker set下所有的broker地址(ip:port)
                            this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs());
                        }
                        // Update Pub info
                        {
                            // 将从namesrv获取到的路由信息转换为TopicPublishInfo
                            // 期间会将没有master的broker set的queue信息去除
                            TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData);
    // 省略中间代码...
    } catch (InterruptedException e) {
        log.warn("updateTopicRouteInfoFromNameServer Exception", e);
    }

    return false;
}

// org.apache.rocketmq.client.impl.factory.MQClientInstance#findBrokerAddressInPublish
public String findBrokerAddressInPublish(final String brokerName) {
    // 上面updateTopicRouteInfoFromNameServer方法将broker set下的broker地址信息保存到brokerAddrTable
    // 再次重申：一个broker set下的broker的brokerName相同
    HashMap<Long/* brokerId */, String/* address */> map = this.brokerAddrTable.get(brokerName);
    if (map != null && !map.isEmpty()) {
        // 没有花样，就是直接返回brokerId时MixAll.MASTER_ID的broker的ip:port信息
        // 前面说过master的brokerId就是MixAll.MASTER_ID，所以获取到的broker是broker set中的master
        return map.get(MixAll.MASTER_ID);
    }

    return null;
}