RocketMQ 源码分析 RouteInfoManager（四）

前言

在上一章分析了NamesrvController的构造函数时，会生成一个RouteInfoManager对象，该对象存放着整个消息集群的相关消息，所以这里单独拿出来分析。其实试想一下namesrv的功能不就是一个提供了通信功能的一个队列嘛，而RouteInfoManager保留了所有信息的路由。所以要想弄明白RocketMQ，RouteInfoManager必须要攻下。

RouteInfoManager的构造函数

主要提供了topic、broker、cluster、liveBroker的路由信息。

public class RouteInfoManager {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
    // 读写锁 
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    // Topic,以及对应的队列信息 
    private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
    // 以Broker Name为单位的Broker集合 
    private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
    // 集群以及属于该集群的Broker列表 
    private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
    // 存活的Broker地址列表 
    private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
    // Broker对应的Filter Server列表 
    private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

    public RouteInfoManager() {
        this.topicQueueTable = new HashMap<String, List<QueueData>>(1024);
        this.brokerAddrTable = new HashMap<String, BrokerData>(128);
        this.clusterAddrTable = new HashMap<String, Set<String>>(32);
        this.brokerLiveTable = new HashMap<String, BrokerLiveInfo>(256);
        this.filterServerTable = new HashMap<String, List<String>>(256);
    }
    ..........省略
}

topicQueueTable

private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;

存放了Topic的一个hashMap，在RocketMQ的插件中RocketMQ控制台，可以看到Topic的列表，这个就是topicQueueTable的一个可视化：

查看QueueData的类信息：

public class QueueData implements Comparable<QueueData> {
    // 队列所属的Broker名称 
    private String brokerName;
    // 读队列数量 
    private int readQueueNums;
    // 写队列数量 
    private int writeQueueNums;
    // Topic的读写权限(2是写 4是读 6是读写) 
    private int perm;
    private int topicSynFlag;
}

点开一个一个topic的配置可以看到和上面类对应的变量：

clusterAddrTable

private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;

根据一个集群名，获得对应的一组BrokerName的列表，在可视化界面就是如下：

虽然只得到了BrokerName，只要调用对应的接口就能得到Broker对应的属性值。

brokerAddrTable

private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;

以Broker Name为Key，BrokerData为Value：

public class BrokerData implements Comparable<BrokerData> {
    private String cluster;
    private String brokerName;
    //同一个brokerName下可以有一个Master和多个Slave,所以brokerAddrs是一个集合 
    private HashMap<Long/* brokerId */, String/* broker address */> brokerAddrs;
    private final Random random = new Random();
}

Master与Slave 的对应关系通过指定相同的BrokerName，不同的BrokerId来定 义，BrokerId为0 表示Master，非0 表示Slave。

brokerLiveTable

private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;

用于存放存活的Broker信息，BrokerLiveInfo：

class BrokerLiveInfo {
    // 最后一次更新时间 
    private long lastUpdateTimestamp;
    // 版本号 
    private DataVersion dataVersion;
    // Netty的Channel 
    private Channel channel;
    /** 
    * HA Broker的地址 
    * 是Slave从Master拉取数据时链接的地址,由brokerIp2+HA端口构成 */ 
    private String haServerAddr;   
}

registerBroker

上面的类的解释可能还存在很多疑惑，下面以registerBroker方法，来实际操作一下上面提到类，看看注册一个Broker发生了什么：

public RegisterBrokerResult registerBroker(
    final String clusterName,
    final String brokerAddr,
    final String brokerName,
    final long brokerId,
    final String haServerAddr,
    //TopicConfigSerializeWrapper比较复杂的数据结构，主要包含了broker上所有的topic信息
    final TopicConfigSerializeWrapper topicConfigWrapper,
    final List<String> filterServerList,
    final Channel channel) {
    RegisterBrokerResult result = new RegisterBrokerResult();
    try {
        try {
            this.lock.writeLock().lockInterruptibly();

            // 更新集群信息（根据集群名字，获取当前集群下面的所有brokerName）
            Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
            // 如果当前集群下面brokerNames为空，则将当前请求broker加入到clusterAddrTable中
            if (null == brokerNames) {
                brokerNames = new HashSet<String>();
                this.clusterAddrTable.put(clusterName, brokerNames);
            }
            brokerNames.add(brokerName);

            boolean registerFirst = false;

            //获取所有的名称为brokerNamed 的brokerData，brokerData保留着了。也就是说同一个BrokerName，有可能有多条的brokerId和broker address与之对应
            // HashMap<Long/* brokerId */, String/* broker address */> brokerAddrs
            BrokerData brokerData = this.brokerAddrTable.get(brokerName);
            if (null == brokerData) {
                registerFirst = true;
                brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());
                this.brokerAddrTable.put(brokerName, brokerData);
            }
            String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);
            registerFirst = registerFirst || (null == oldAddr);

            // 更新Topic信息
            //如果topicConfigWrapper不为空，且当前brokerId == 0，即为当前broker为master
            //这里会判断只有master才会创建QueueData，因为只有master才包含了读写队列的信息
            if (null != topicConfigWrapper
                && MixAll.MASTER_ID == brokerId) {
                // 如果Topic配置信息发生变更或者该broker为第一次注册
                if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())
                    || registerFirst) {
                    // 获取所有topic信息
                    ConcurrentMap<String, TopicConfig> tcTable =
                        topicConfigWrapper.getTopicConfigTable();
                    if (tcTable != null) {
                        // 遍历所有Topic
                        for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) {
                            // 根据brokername及topicconfig（read、write queue数量等）新增或者更新到topicQueueTable中
                            this.createAndUpdateQueueData(brokerName, entry.getValue());
                        }
                    }
                }
            }

            // 更新最后变更时间（将brokerLiveTable中保存的对应的broker的更新时间戳，设置为当前时间）
            BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddr,
                new BrokerLiveInfo(
                    System.currentTimeMillis(),
                    topicConfigWrapper.getDataVersion(),
                    channel,
                    haServerAddr));
            if (null == prevBrokerLiveInfo) {
                log.info("new broker registered, {} HAServer: {}", brokerAddr, haServerAddr);
            }

            if (filterServerList != null) {
                if (filterServerList.isEmpty()) {
                    this.filterServerTable.remove(brokerAddr);
                } else {
                    this.filterServerTable.put(brokerAddr, filterServerList);
                }
            }

            // 返回值（如果当前broker为slave节点）则将haServerAddr、masterAddr等信息设置到result返回值中
            if (MixAll.MASTER_ID != brokerId) {
                // 通过brokename的brokedate获取当前slave节点的master节点addr
                String masterAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
                if (masterAddr != null) {
                    BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.get(masterAddr);
                    if (brokerLiveInfo != null) {
                        result.setHaServerAddr(brokerLiveInfo.getHaServerAddr());
                        result.setMasterAddr(masterAddr);
                    }
                }
            }
        } finally {
            this.lock.writeLock().unlock();
        }
    } catch (Exception e) {
        log.error("registerBroker Exception", e);
    }

    return result;
}

插入过程可以看如下这张图：

疑惑

1. topicQueueTable有一点不明白，为什么一个topic对应的是一个List，难道不是一个topic对应一个QueueData吗？难道可以存在相同的topic？

   因为每一个topic可以存在不同的broker中，不同的broker就有不同的QueueData。

2. brokerAddrTable根据一个brokername获得一个BrokerData，可是为什么一个BrokerData中有一个hashMap的brokerAddrs呢？我能理解的是一个brokername，会有多个address和id，但是cluster为什么又是同一个呢？

   这是因为你理解错了，以为BrokerName是唯一的，实际上是指定多个broker为一个name，只不过指定了不同的id和addres来区分是master还是slave。