深入剖析RocketMQ源码-NameServer

public static NamesrvController main0(String[] args) {
    try {        //创建namesrvController        NamesrvController controller = createNamesrvController(args);        //初始化并启动NamesrvController        start(controller);        String tip = "The Name Server boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();        log.info(tip);        System.out.printf("%s%n", tip);        return controller;    } catch (Throwable e) {        e.printStackTrace();        System.exit(-1);    }
    return null;}

public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {    // 设置版本号为当前版本号    System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));    //PackageConflictDetect.detectFastjson();  //构造org.apache.commons.cli.Options,并添加-h -n参数，-h参数是打印帮助信息，-n参数是指定namesrvAddr    Options options = ServerUtil.buildCommandlineOptions(new Options());    //初始化commandLine，并在options中添加-c -p参数，-c指定nameserver的配置文件路径，-p标识打印配置信息    commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());    if (null == commandLine) {        System.exit(-1);        return null;    }  //nameserver配置类，业务参数    final NamesrvConfig namesrvConfig = new NamesrvConfig();    //netty服务器配置类，网络参数    final NettyServerConfig nettyServerConfig = new NettyServerConfig();    //设置nameserver的端口号    nettyServerConfig.setListenPort(9876);    //命令带有-c参数，说明指定配置文件，需要根据配置文件路径读取配置文件内容，并将文件中配置信息赋值给NamesrvConfig和NettyServerConfig    if (commandLine.hasOption('c')) {        String file = commandLine.getOptionValue('c');        if (file != null) {            InputStream in = new BufferedInputStream(new FileInputStream(file));            properties = new Properties();            properties.load(in);            //反射的方式            MixAll.properties2Object(properties, namesrvConfig);            MixAll.properties2Object(properties, nettyServerConfig);      //设置配置文件路径            namesrvConfig.setConfigStorePath(file);
            System.out.printf("load config properties file OK, %s%n", file);            in.close();        }    }  //命令行带有-p，说明是打印参数的命令，那么就打印出NamesrvConfig和NettyServerConfig的属性。在启动NameServer时可以先使用./mqnameserver -c configFile -p打印当前加载的配置属性     if (commandLine.hasOption('p')) {        InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);        MixAll.printObjectProperties(console, namesrvConfig);        MixAll.printObjectProperties(console, nettyServerConfig);        //打印参数命令不需要启动nameserver服务，只需要打印参数即可        System.exit(0);    }  //解析命令行参数，并加载到namesrvConfig中    MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);  //检查ROCKETMQ_HOME，不能为空    if (null == namesrvConfig.getRocketmqHome()) {        System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation%n", MixAll.ROCKETMQ_HOME_ENV);        System.exit(-2);    }  //初始化logback日志工厂，rocketmq默认使用logback作为日志输出    LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();    JoranConfigurator configurator = new JoranConfigurator();    configurator.setContext(lc);    lc.reset();    configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");
    log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    MixAll.printObjectProperties(log, namesrvConfig);    MixAll.printObjectProperties(log, nettyServerConfig);  //创建NamesrvController    final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
    //将全局Properties的内容复制到NamesrvController.Configuration.allConfigs中    // remember all configs to prevent discard    controller.getConfiguration().registerConfig(properties);
    return controller;}

public boolean initialize() {  //加载kvConfigPath下kvConfig.json配置文件里的KV配置，然后将这些配置放到KVConfigManager#configTable属性中    this.kvConfigManager.load();  //根据nettyServerConfig初始化一个netty服务器。    //brokerHousekeepingService是在NamesrvController实例化时构造函数里实例化的，该类负责Broker连接事件的处理，实现了ChannelEventListener，主要用来管理RouteInfoManager的brokerLiveTable    this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);  //初始化负责处理Netty网络交互数据的线程池，默认线程数是8个    this.remotingExecutor =        Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));  //注册Netty服务端业务处理逻辑，如果开启了clusterTest，那么注册的请求处理类是ClusterTestRequestProcessor，否则请求处理类是DefaultRequestProcessor    this.registerProcessor();  //注册心跳机制线程池，延迟5秒启动，每隔10秒遍历RouteInfoManager#brokerLiveTable这个属性，用来扫描不存活的broker    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        @Override        public void run() {            NamesrvController.this.routeInfoManager.scanNotActiveBroker();        }    }, 5, 10, TimeUnit.SECONDS);  //注册打印KV配置线程池，延迟1分钟启动、每10分钟打印出kvConfig配置    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        @Override        public void run() {            NamesrvController.this.kvConfigManager.printAllPeriodically();        }    }, 1, 10, TimeUnit.MINUTES);  //rocketmq可以通过开启TLS来提高数据传输的安全性，如果开启了，那么需要注册一个监听器来重新加载SslContext    if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {        // Register a listener to reload SslContext        try {            fileWatchService = new FileWatchService(                new String[] {                    TlsSystemConfig.tlsServerCertPath,                    TlsSystemConfig.tlsServerKeyPath,                    TlsSystemConfig.tlsServerTrustCertPath                },                new FileWatchService.Listener() {                    boolean certChanged, keyChanged = false;                    @Override                    public void onChanged(String path) {                        if (path.equals(TlsSystemConfig.tlsServerTrustCertPath)) {                            log.info("The trust certificate changed, reload the ssl context");                            reloadServerSslContext();                        }                        if (path.equals(TlsSystemConfig.tlsServerCertPath)) {                            certChanged = true;                        }                        if (path.equals(TlsSystemConfig.tlsServerKeyPath)) {                            keyChanged = true;                        }                        if (certChanged && keyChanged) {                            log.info("The certificate and private key changed, reload the ssl context");                            certChanged = keyChanged = false;                            reloadServerSslContext();                        }                    }                    private void reloadServerSslContext() {                        ((NettyRemotingServer) remotingServer).loadSslContext();                    }                });        } catch (Exception e) {            log.warn("FileWatchService created error, can't load the certificate dynamically");        }    }
    return true;}

Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {    @Override    public Void call() throws Exception {        controller.shutdown();        return null;    }}));

public void start() throws Exception {    //启动netty服务    this.remotingServer.start();  //如果开启了TLS    if (this.fileWatchService != null) {        this.fileWatchService.start();    }}

private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

topicQueueTable:{    "topic1": [        {            "brokerName": "broker-a",            "readQueueNums":4,            "writeQueueNums":4,            "perm":6,            "topicSynFlag":0,        },        {            "brokerName": "broker-b",            "readQueueNums":4,            "writeQueueNums":4,            "perm":6,            "topicSynFlag":0,        }    ]}

brokerAddrTable:{    "broker-a": {        "cluster": "c1",        "brokerName": "broker-a",        "brokerAddrs": {            0: "192.168.1.1:10000",            1: "192.168.1.2:10000"        }    },    "broker-b": {        "cluster": "c1",        "brokerName": "broker-b",        "brokerAddrs": {            0: "192.168.1.3:10000",            1: "192.168.1.4:10000"        }    }}

clusterAddrTable:{    "c1": ["broker-a","broker-b"]}

brokerLiveTable:{    "192.168.1.1:10000": {            "lastUpdateTimestamp": 1518270318980,            "dataVersion":versionObj1,            "channel":channelObj,            "haServerAddr":""    },    "192.168.1.2:10000": {            "lastUpdateTimestamp": 1518270318980,            "dataVersion":versionObj1,            "channel":channelObj,            "haServerAddr":"192.168.1.1:10000"     },    "192.168.1.3:10000": {            "lastUpdateTimestamp": 1518270318980,            "dataVersion":versionObj1,            "channel":channelObj,            "haServerAddr":""     },    "192.168.1.4:10000": {            "lastUpdateTimestamp": 1518270318980,            "dataVersion":versionObj1,            "channel":channelObj,            "haServerAddr":"192.168.1.3:10000"     }}

this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    @Override    public void run() {        try {            BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());        } catch (Throwable e) {            log.error("registerBrokerAll Exception", e);        }    }}, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);

public List<RegisterBrokerResult> registerBrokerAll(    final String clusterName,    final String brokerAddr,    final String brokerName,    final long brokerId,    final String haServerAddr,    final TopicConfigSerializeWrapper topicConfigWrapper,    final List<String> filterServerList,    final boolean oneway,    final int timeoutMills,    final boolean compressed) {
    final List<RegisterBrokerResult> registerBrokerResultList = new CopyOnWriteArrayList<>();    //获取nameserver地址列表    List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();    if (nameServerAddressList != null && nameServerAddressList.size() > 0) {    /**      *封装请求包头start      *封装请求包头，主要封装broker相关信息    **/        final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();        requestHeader.setBrokerAddr(brokerAddr);        requestHeader.setBrokerId(brokerId);        requestHeader.setBrokerName(brokerName);        requestHeader.setClusterName(clusterName);        requestHeader.setHaServerAddr(haServerAddr);        requestHeader.setCompressed(compressed);    //封装requestBody，包括topic和filterServerList相关信息        RegisterBrokerBody requestBody = new RegisterBrokerBody();        requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);        requestBody.setFilterServerList(filterServerList);        final byte[] body = requestBody.encode(compressed);        final int bodyCrc32 = UtilAll.crc32(body);        requestHeader.setBodyCrc32(bodyCrc32);        /**      *封装请求包头end    **/        //开启多线程到每个nameserver进行注册        final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());        for (final String namesrvAddr : nameServerAddressList) {            brokerOuterExecutor.execute(new Runnable() {                @Override                public void run() {                    try {                        //实际进行注册方法                        RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);                        if (result != null) {                            //封装nameserver返回的信息                            registerBrokerResultList.add(result);                        }
                        log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);                    } catch (Exception e) {                        log.warn("registerBroker Exception, {}", namesrvAddr, e);                    } finally {                        countDownLatch.countDown();                    }                }            });        }
        try {            countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);        } catch (InterruptedException e) {        }    }
    return registerBrokerResultList;}

private RegisterBrokerResult registerBroker(    final String namesrvAddr,    final boolean oneway,    final int timeoutMills,    final RegisterBrokerRequestHeader requestHeader,    final byte[] body) throws RemotingCommandException, MQBrokerException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,InterruptedException {    //创建请求指令，需要注意RequestCode.REGISTER_BROKER，nameserver端的网络处理器会根据requestCode进行相应的业务处理    RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.REGISTER_BROKER, requestHeader);    request.setBody(body);  //基于netty进行网络传输    if (oneway) {        //如果是单向调用，没有返回值，不返回nameserver返回结果        try {            this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);        } catch (RemotingTooMuchRequestException e) {            // Ignore        }        return null;    }  //异步调用向nameserver发起注册，获取nameserver的返回信息    RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);    assert response != null;    switch (response.getCode()) {        case ResponseCode.SUCCESS: {            //获取返回的reponseHeader            RegisterBrokerResponseHeader responseHeader =                (RegisterBrokerResponseHeader) response.decodeCommandCustomHeader(RegisterBrokerResponseHeader.class);            //重新封装返回结果，更新masterAddr和haServerAddr            RegisterBrokerResult result = new RegisterBrokerResult();            result.setMasterAddr(responseHeader.getMasterAddr());            result.setHaServerAddr(responseHeader.getHaServerAddr());            if (response.getBody() != null) {                result.setKvTable(KVTable.decode(response.getBody(), KVTable.class));            }            return result;        }        default:            break;    }
    throw new MQBrokerException(response.getCode(), response.getRemark(), requestHeader == null ? null : requestHeader.getBrokerAddr());}

public RemotingCommand processRequest(ChannelHandlerContext ctx,        RemotingCommand request) throws RemotingCommandException {    if (ctx != null) {        log.debug("receive request, {} {} {}",                  request.getCode(),                  RemotingHelper.parseChannelRemoteAddr(ctx.channel()),                  request);    }    switch (request.getCode()) {        ......        //，如果是RequestCode.REGISTER_BROKER，进行broker注册        case RequestCode.REGISTER_BROKER:            Version brokerVersion = MQVersion.value2Version(request.getVersion());            if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) {                return this.registerBrokerWithFilterServer(ctx, request);            } else {                return this.registerBroker(ctx, request);            }        ......        default:            break;    }    return null;}

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 {            //加写锁，防止并发写RoutInfoManager中的路由表信息。            this.lock.writeLock().lockInterruptibly();      //根据clusterName从clusterAddrTable中获取所有broker名字集合            Set<String> brokerNames = this.clusterAddrTable.get(clusterName);            //如果没有获取到，说明broker所属集群还没记录，那么需要创建，并将brokerName加入到集群的broker集合中            if (null == brokerNames) {                brokerNames = new HashSet<String>();                this.clusterAddrTable.put(clusterName, brokerNames);            }            brokerNames.add(brokerName);                  boolean registerFirst = false;      //根据brokerName尝试从brokerAddrTable中获取brokerData            BrokerData brokerData = this.brokerAddrTable.get(brokerName);            if (null == brokerData) {                //如果没获取到brokerData，新建BrokerData并放入brokerAddrTable，registerFirst设为true；                registerFirst = true;                brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());                this.brokerAddrTable.put(brokerName, brokerData);            }            //更新brokerData中的brokerAddrs            Map<Long, String> brokerAddrsMap = brokerData.getBrokerAddrs();            //考虑到可能出现master挂了，slave变成master的情况，这时候brokerId会变成0，这时候需要把老的brokerAddr给删除            //Switch slave to master: first remove <1, IP:PORT> in namesrv, then add <0, IP:PORT>            //The same IP:PORT must only have one record in brokerAddrTable            Iterator<Entry<Long, String>> it = brokerAddrsMap.entrySet().iterator();            while (it.hasNext()) {                Entry<Long, String> item = it.next();                if (null != brokerAddr && brokerAddr.equals(item.getValue()) && brokerId != item.getKey()) {                    it.remove();                }            }      //更新brokerAddrs，根据返回的oldAddr判断是否是第一次注册的broker            String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);            registerFirst = registerFirst || (null == oldAddr);
            //如过Broker是Master，并且Broker的Topic配置信息发生变化或者是首次注册，需要创建或更新Topic路由元数据，填充topicQueueTable            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()) {                            //创建或更新Topic路由元数据                            this.createAndUpdateQueueData(brokerName, entry.getValue());                        }                    }                }            }      //更新BrokerLivelnfo,BrokeLivelnfo是执行路由删除的重要依据            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);            }      //注册Broker的filterServer地址列表            if (filterServerList != null) {                if (filterServerList.isEmpty()) {                    this.filterServerTable.remove(brokerAddr);                } else {                    this.filterServerTable.put(brokerAddr, filterServerList);                }            }      //如果此Broker为从节点，则需要查找Broker Master的节点信息，并更新对应masterAddr属性            if (MixAll.MASTER_ID != brokerId) {                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;}

public void onChannelDestroy(String remoteAddr, Channel channel) {    String brokerAddrFound = null;    if (channel != null) {        try {            try {                //加读锁                this.lock.readLock().lockInterruptibly();                //通过channel从brokerLiveTable中找出对应的Broker地址                Iterator<Entry<String, BrokerLiveInfo>> itBrokerLiveTable =                    this.brokerLiveTable.entrySet().iterator();                while (itBrokerLiveTable.hasNext()) {                    Entry<String, BrokerLiveInfo> entry = itBrokerLiveTable.next();                    if (entry.getValue().getChannel() == channel) {                        brokerAddrFound = entry.getKey();                        break;                    }                }            } finally {                //释放读锁                this.lock.readLock().unlock();            }        } catch (Exception e) {            log.error("onChannelDestroy Exception", e);        }    }  //若该Broker已经从存活的Broker地址列表中被清除，则直接使用remoteAddr    if (null == brokerAddrFound) {        brokerAddrFound = remoteAddr;    } else {        log.info("the broker's channel destroyed, {}, clean it's data structure at once", brokerAddrFound);    }
    if (brokerAddrFound != null && brokerAddrFound.length() > 0) {
        try {            try {                //申请写锁                this.lock.writeLock().lockInterruptibly();                //根据brokerAddress，将这个brokerAddress从brokerLiveTable和filterServerTable中移除                this.brokerLiveTable.remove(brokerAddrFound);                this.filterServerTable.remove(brokerAddrFound);                String brokerNameFound = null;                boolean removeBrokerName = false;                Iterator<Entry<String, BrokerData>> itBrokerAddrTable =                    this.brokerAddrTable.entrySet().iterator();                //遍历brokerAddrTable                while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) {                    BrokerData brokerData = itBrokerAddrTable.next().getValue();
                    Iterator<Entry<Long, String>> it = brokerData.getBrokerAddrs().entrySet().iterator();                    while (it.hasNext()) {                        Entry<Long, String> entry = it.next();                        Long brokerId = entry.getKey();                        String brokerAddr = entry.getValue();                        //根据brokerAddress找到对应的brokerData，并将brokerData中对应的brokerAddress移除                        if (brokerAddr.equals(brokerAddrFound)) {                            brokerNameFound = brokerData.getBrokerName();                            it.remove();                            log.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",                                     brokerId, brokerAddr);                            break;                        }                    }          //如果移除后，整个brokerData的brokerAddress空了，那么将整个brokerData移除                    if (brokerData.getBrokerAddrs().isEmpty()) {                        removeBrokerName = true;                        itBrokerAddrTable.remove();                        log.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",                                 brokerData.getBrokerName());                    }                }
                if (brokerNameFound != null && removeBrokerName) {                    //遍历clusterAddrTable                    Iterator<Entry<String, Set<String>>> it = this.clusterAddrTable.entrySet().iterator();                    while (it.hasNext()) {                        Entry<String, Set<String>> entry = it.next();                        String clusterName = entry.getKey();                        Set<String> brokerNames = entry.getValue();                        //根据第三步中获取的需要移除的brokerName，将对应的brokerName移除了                        boolean removed = brokerNames.remove(brokerNameFound);                        if (removed) {                            log.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",                                     brokerNameFound, clusterName);              //如果移除后，该集合为空，那么将整个集群从clusterAddrTable中移除                            if (brokerNames.isEmpty()) {                                log.info("remove the clusterName[{}] from clusterAddrTable, because channel destroyed and no broker in this cluster",                                         clusterName);                                it.remove();                            }
                            break;                        }                    }                }
                if (removeBrokerName) {                    Iterator<Entry<String, List<QueueData>>> itTopicQueueTable =                        this.topicQueueTable.entrySet().iterator();                    //遍历topicQueueTable                    while (itTopicQueueTable.hasNext()) {                        Entry<String, List<QueueData>> entry = itTopicQueueTable.next();                        String topic = entry.getKey();                        List<QueueData> queueDataList = entry.getValue();
                        Iterator<QueueData> itQueueData = queueDataList.iterator();                        while (itQueueData.hasNext()) {                            QueueData queueData = itQueueData.next();                            //根据brokerName，将topic下对应的broker移除掉                            if (queueData.getBrokerName().equals(brokerNameFound)) {                                itQueueData.remove();                                log.info("remove topic[{} {}], from topicQueueTable, because channel destroyed",                                         topic, queueData);                            }                        }            //如果该topic下只有一个待移除的broker，那么该topic也从table中移除                        if (queueDataList.isEmpty()) {                            itTopicQueueTable.remove();                            log.info("remove topic[{}] all queue, from topicQueueTable, because channel destroyed",                                     topic);                        }                    }                }            } finally {                //释放写锁                this.lock.writeLock().unlock();            }        } catch (Exception e) {            log.error("onChannelDestroy Exception", e);        }    }}

private void startScheduledTask() {    ......    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        @Override        public void run() {            try {                //从nameserver更新最新的topic路由信息                MQClientInstance.this.updateTopicRouteInfoFromNameServer();            } catch (Exception e) {                log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e);            }        }    }, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);
    ......}
/**    * 从nameserver获取topic路由信息    */public TopicRouteData getTopicRouteInfoFromNameServer(final String topic, final long timeoutMillis,                                                      boolean allowTopicNotExist) throws MQClientException, InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {    ......    //向nameserver发送请求包，requestCode为RequestCode.GET_ROUTEINFO_BY_TOPIC    RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ROUTEINFO_BY_TOPIC, requestHeader);  ......}

public TopicRouteData pickupTopicRouteData(final String topic) {    TopicRouteData topicRouteData = new TopicRouteData();    boolean foundQueueData = false;    boolean foundBrokerData = false;    Set<String> brokerNameSet = new HashSet<String>();    List<BrokerData> brokerDataList = new LinkedList<BrokerData>();    topicRouteData.setBrokerDatas(brokerDataList);
    HashMap<String, List<String>> filterServerMap = new HashMap<String, List<String>>();    topicRouteData.setFilterServerTable(filterServerMap);
    try {        try {            //加读锁            this.lock.readLock().lockInterruptibly();            //从元数据topicQueueTable中根据topic名字获取队列集合            List<QueueData> queueDataList = this.topicQueueTable.get(topic);            if (queueDataList != null) {                //将获取到的队列集合写入topicRouteData的queueDatas中                topicRouteData.setQueueDatas(queueDataList);                foundQueueData = true;
                Iterator<QueueData> it = queueDataList.iterator();                while (it.hasNext()) {                    QueueData qd = it.next();                    brokerNameSet.add(qd.getBrokerName());                }        //遍历从QueueData集合中提取的brokerName                for (String brokerName : brokerNameSet) {                    //根据brokerName从brokerAddrTable获取brokerData                    BrokerData brokerData = this.brokerAddrTable.get(brokerName);                    if (null != brokerData) {                        //克隆brokerData对象，并写入到topicRouteData的brokerDatas中                        BrokerData brokerDataClone = new BrokerData(brokerData.getCluster(), brokerData.getBrokerName(), (HashMap<Long, String>) brokerData.getBrokerAddrs().clone());                        brokerDataList.add(brokerDataClone);                        foundBrokerData = true;                        //遍历brokerAddrs                        for (final String brokerAddr : brokerDataClone.getBrokerAddrs().values()) {                            //根据brokerAddr获取filterServerList，封装后写入到topicRouteData的filterServerTable中                            List<String> filterServerList = this.filterServerTable.get(brokerAddr);                            filterServerMap.put(brokerAddr, filterServerList);                        }                    }                }            }        } finally {            //释放读锁            this.lock.readLock().unlock();        }    } catch (Exception e) {        log.error("pickupTopicRouteData Exception", e);    }
    log.debug("pickupTopicRouteData {} {}", topic, topicRouteData);
    if (foundBrokerData && foundQueueData) {        return topicRouteData;    }
    return null;}

public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx,                                           RemotingCommand request) throws RemotingCommandException {    ......  //这块代码就是上面解析的代码，获取到topicRouteData对象    TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());
    if (topicRouteData != null) {        //判断nameserver的orderMessageEnable配置是否打开        if (this.namesrvController.getNamesrvConfig().isOrderMessageEnable()) {            //如果配置打开了，根据namespace和topic名字获取kvConfig配置文件中顺序消息配置内容            String orderTopicConf =                this.namesrvController.getKvConfigManager().getKVConfig(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG,                                                                        requestHeader.getTopic());            //封装orderTopicConf            topicRouteData.setOrderTopicConf(orderTopicConf);        }
        byte[] content = topicRouteData.encode();        response.setBody(content);        response.setCode(ResponseCode.SUCCESS);        response.setRemark(null);        return response;    }  //如果没有获取到topic路由，那么reponseCode为TOPIC_NOT_EXIST    response.setCode(ResponseCode.TOPIC_NOT_EXIST);    response.setRemark("No topic route info in name server for the topic: " + requestHeader.getTopic()                       + FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));    return response;}