0%

# 聊一下 RocketMQ 的 NameServer 源码

Posted on 2020-07-05 In MQ , RocketMQ , 消息队列 , RocketMQ , 中间件 , RocketMQ Views: Views: Disqus:

前面介绍了，nameserver相当于dubbo的注册中心，用与管理broker，broker会在启动的时候注册到nameserver，并且会发送心跳给namaserver，nameserver负责保存活跃的broker，包括master和slave，同时保存topic和topic下的队列，以及filter列表，然后为producer和consumer的请求提供服务。

### 启动过程

    public static void main(String[] args) {
            main0(args);
        }

        public static NamesrvController main0(String[] args) {

            try {
                NamesrvController controller = createNamesrvController(args);
                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;
        }

入口的代码时这样子，其实主要的逻辑在createNamesrvController和start方法，来看下这两个的实现

    public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {
            System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
            //PackageConflictDetect.detectFastjson();

            Options options = ServerUtil.buildCommandlineOptions(new Options());
            commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());
            if (null == commandLine) {
                System.exit(-1);
                return null;
            }

            final NamesrvConfig namesrvConfig = new NamesrvConfig();
            final NettyServerConfig nettyServerConfig = new NettyServerConfig();
            nettyServerConfig.setListenPort(9876);
            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();
                }
            }

            if (commandLine.hasOption('p')) {
                InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
                MixAll.printObjectProperties(console, namesrvConfig);
                MixAll.printObjectProperties(console, nettyServerConfig);
                System.exit(0);
            }

            MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);

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

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

            final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);

            // remember all configs to prevent discard
            controller.getConfiguration().registerConfig(properties);

            return controller;
        }

这个方法里其实主要是读取一些配置啥的，不是很复杂，

    public static NamesrvController start(final NamesrvController controller) throws Exception {

            if (null == controller) {
                throw new IllegalArgumentException("NamesrvController is null");
            }

            boolean initResult = controller.initialize();
            if (!initResult) {
                controller.shutdown();
                System.exit(-3);
            }

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

            controller.start();

            return controller;
        }

这个start里主要关注initialize方法，后面就是一个停机的hook，来看下initialize方法

    public boolean initialize() {

            this.kvConfigManager.load();

            this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);

            this.remotingExecutor =
                Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));

            this.registerProcessor();

            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

                @Override
                public void run() {
                    NamesrvController.this.routeInfoManager.scanNotActiveBroker();
                }
            }, 5, 10, TimeUnit.SECONDS);

            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

                @Override
                public void run() {
                    NamesrvController.this.kvConfigManager.printAllPeriodically();
                }
            }, 1, 10, TimeUnit.MINUTES);

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

这里的kvConfigManager主要是来加载NameServer的配置参数，存到org.apache.rocketmq.namesrv.kvconfig.KVConfigManager#configTable中，然后是以BrokerHousekeepingService对象为参数初始化NettyRemotingServer对象，BrokerHousekeepingService对象作为该Netty连接中Socket链接的监听器（ChannelEventListener）；监听与Broker建立的渠道的状态（空闲、关闭、异常三个状态），并调用BrokerHousekeepingService的相应onChannel方法。其中渠道的空闲、关闭、异常状态均调用RouteInfoManager.onChannelDestory方法处理。这个BrokerHousekeepingService可以字面化地理解为broker的管家服务，这个类内部三个状态方法其实都是调用的org.apache.rocketmq.namesrv.NamesrvController#getRouteInfoManager方法，而这个RouteInfoManager里面的对象有这些

    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与queue的对应关系
        private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
      // Broker名称与broker属性的map
        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地址与过滤器
        private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

然后接下去就是初始化了一个线程池，然后注册默认的处理类`this.registerProcessor();`默认都是这个处理器去处理请求 `org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor#DefaultRequestProcessor`然后是初始化两个定时任务

第一是每10秒检查一遍Broker的状态的定时任务，调用scanNotActiveBroker方法；遍历brokerLiveTable集合，查看每个broker的最后更新时间（BrokerLiveInfo.lastUpdateTimestamp）是否超过2分钟，若超过则关闭该broker的渠道并调用RouteInfoManager.onChannelDestory方法清理RouteInfoManager类的topicQueueTable、brokerAddrTable、clusterAddrTable、filterServerTable成员变量。

    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

                @Override
                public void run() {
                    NamesrvController.this.routeInfoManager.scanNotActiveBroker();
                }
            }, 5, 10, TimeUnit.SECONDS);
    public void scanNotActiveBroker() {
            Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
            while (it.hasNext()) {
                Entry<String, BrokerLiveInfo> next = it.next();
                long last = next.getValue().getLastUpdateTimestamp();
                if ((last + BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) {
                    RemotingUtil.closeChannel(next.getValue().getChannel());
                    it.remove();
                    log.warn("The broker channel expired, {} {}ms", next.getKey(), BROKER_CHANNEL_EXPIRED_TIME);
                    this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
                }
            }
        }

        public void onChannelDestroy(String remoteAddr, Channel channel) {
            String brokerAddrFound = null;
            if (channel != null) {
                try {
                    try {
                        this.lock.readLock().lockInterruptibly();
                        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);
                }
            }

            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();
                        this.brokerLiveTable.remove(brokerAddrFound);
                        this.filterServerTable.remove(brokerAddrFound);
                        String brokerNameFound = null;
                        boolean removeBrokerName = false;
                        Iterator<Entry<String, BrokerData>> itBrokerAddrTable =
                            this.brokerAddrTable.entrySet().iterator();
                        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();
                                if (brokerAddr.equals(brokerAddrFound)) {
                                    brokerNameFound = brokerData.getBrokerName();
                                    it.remove();
                                    log.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",
                                        brokerId, brokerAddr);
                                    break;
                                }
                            }

                            if (brokerData.getBrokerAddrs().isEmpty()) {
                                removeBrokerName = true;
                                itBrokerAddrTable.remove();
                                log.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",
                                    brokerData.getBrokerName());
                            }
                        }

                        if (brokerNameFound != null && removeBrokerName) {
                            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();
                                boolean removed = brokerNames.remove(brokerNameFound);
                                if (removed) {
                                    log.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",
                                        brokerNameFound, clusterName);

                                    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();
                            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();
                                    if (queueData.getBrokerName().equals(brokerNameFound)) {
                                        itQueueData.remove();
                                        log.info("remove topic[{} {}], from topicQueueTable, because channel destroyed",
                                            topic, queueData);
                                    }
                                }

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

第二个是每10分钟打印一次NameServer的配置参数。即KVConfigManager.configTable变量的内容。

    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

                @Override
                public void run() {
                    NamesrvController.this.kvConfigManager.printAllPeriodically();
                }
            }, 1, 10, TimeUnit.MINUTES);

然后这个初始化就差不多完成了，后面只需要把remotingServer start一下就好了

### 处理请求

直接上代码，其实主体是swtich case去判断

    @Override
        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()) {
                case RequestCode.PUT_KV_CONFIG:
                    return this.putKVConfig(ctx, request);
                case RequestCode.GET_KV_CONFIG:
                    return this.getKVConfig(ctx, request);
                case RequestCode.DELETE_KV_CONFIG:
                    return this.deleteKVConfig(ctx, request);
                case RequestCode.QUERY_DATA_VERSION:
                    return queryBrokerTopicConfig(ctx, request);
                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);
                    }
                case RequestCode.UNREGISTER_BROKER:
                    return this.unregisterBroker(ctx, request);
                case RequestCode.GET_ROUTEINTO_BY_TOPIC:
                    return this.getRouteInfoByTopic(ctx, request);
                case RequestCode.GET_BROKER_CLUSTER_INFO:
                    return this.getBrokerClusterInfo(ctx, request);
                case RequestCode.WIPE_WRITE_PERM_OF_BROKER:
                    return this.wipeWritePermOfBroker(ctx, request);
                case RequestCode.GET_ALL_TOPIC_LIST_FROM_NAMESERVER:
                    return getAllTopicListFromNameserver(ctx, request);
                case RequestCode.DELETE_TOPIC_IN_NAMESRV:
                    return deleteTopicInNamesrv(ctx, request);
                case RequestCode.GET_KVLIST_BY_NAMESPACE:
                    return this.getKVListByNamespace(ctx, request);
                case RequestCode.GET_TOPICS_BY_CLUSTER:
                    return this.getTopicsByCluster(ctx, request);
                case RequestCode.GET_SYSTEM_TOPIC_LIST_FROM_NS:
                    return this.getSystemTopicListFromNs(ctx, request);
                case RequestCode.GET_UNIT_TOPIC_LIST:
                    return this.getUnitTopicList(ctx, request);
                case RequestCode.GET_HAS_UNIT_SUB_TOPIC_LIST:
                    return this.getHasUnitSubTopicList(ctx, request);
                case RequestCode.GET_HAS_UNIT_SUB_UNUNIT_TOPIC_LIST:
                    return this.getHasUnitSubUnUnitTopicList(ctx, request);
                case RequestCode.UPDATE_NAMESRV_CONFIG:
                    return this.updateConfig(ctx, request);
                case RequestCode.GET_NAMESRV_CONFIG:
                    return this.getConfig(ctx, request);
                default:
                    break;
            }
            return null;
        }

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

做了个简单的版本管理，我们看下前面一个的代码

    public RemotingCommand registerBrokerWithFilterServer(ChannelHandlerContext ctx, RemotingCommand request)
        throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(RegisterBrokerResponseHeader.class);
        final RegisterBrokerResponseHeader responseHeader = (RegisterBrokerResponseHeader) response.readCustomHeader();
        final RegisterBrokerRequestHeader requestHeader =
            (RegisterBrokerRequestHeader) request.decodeCommandCustomHeader(RegisterBrokerRequestHeader.class);

        if (!checksum(ctx, request, requestHeader)) {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("crc32 not match");
            return response;
        }

        RegisterBrokerBody registerBrokerBody = new RegisterBrokerBody();

        if (request.getBody() != null) {
            try {
                registerBrokerBody = RegisterBrokerBody.decode(request.getBody(), requestHeader.isCompressed());
            } catch (Exception e) {
                throw new RemotingCommandException("Failed to decode RegisterBrokerBody", e);
            }
        } else {
            registerBrokerBody.getTopicConfigSerializeWrapper().getDataVersion().setCounter(new AtomicLong(0));
            registerBrokerBody.getTopicConfigSerializeWrapper().getDataVersion().setTimestamp(0);
        }

        RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(
            requestHeader.getClusterName(),
            requestHeader.getBrokerAddr(),
            requestHeader.getBrokerName(),
            requestHeader.getBrokerId(),
            requestHeader.getHaServerAddr(),
            registerBrokerBody.getTopicConfigSerializeWrapper(),
            registerBrokerBody.getFilterServerList(),
            ctx.channel());

        responseHeader.setHaServerAddr(result.getHaServerAddr());
        responseHeader.setMasterAddr(result.getMasterAddr());

        byte[] jsonValue = this.namesrvController.getKvConfigManager().getKVListByNamespace(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG);
        response.setBody(jsonValue);

        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

可以看到主要的逻辑还是在`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 {
                    this.lock.writeLock().lockInterruptibly();

                  // 更新这个clusterAddrTable
                    Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
                    if (null == brokerNames) {
                        brokerNames = new HashSet<String>();
                        this.clusterAddrTable.put(clusterName, brokerNames);
                    }
                    brokerNames.add(brokerName);

                    boolean registerFirst = false;

                  // 更新brokerAddrTable
                    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);
                    }
                    Map<Long, String> brokerAddrsMap = brokerData.getBrokerAddrs();
                    //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();
                        }
                    }

                    String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);
                    registerFirst = registerFirst || (null == oldAddr);

                  // 更新了org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#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()) {
                                    this.createAndUpdateQueueData(brokerName, entry.getValue());
                                }
                            }
                        }
                    }

                  // 更新活跃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);
                    }

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

                  // 当当前broker非master时返回master信息
                    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;
        }

这个是注册 broker 的逻辑，再看下根据 topic 获取 broker 信息和 topic 信息，`org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor#getRouteInfoByTopic` 主要是这个方法的逻辑

    public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx,
            RemotingCommand request) throws RemotingCommandException {
            final RemotingCommand response = RemotingCommand.createResponseCommand(null);
            final GetRouteInfoRequestHeader requestHeader =
                (GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);

            TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());

            if (topicRouteData != null) {
                if (this.namesrvController.getNamesrvConfig().isOrderMessageEnable()) {
                    String orderTopicConf =
                        this.namesrvController.getKvConfigManager().getKVConfig(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG,
                            requestHeader.getTopic());
                    topicRouteData.setOrderTopicConf(orderTopicConf);
                }

                byte[] content = topicRouteData.encode();
                response.setBody(content);
                response.setCode(ResponseCode.SUCCESS);
                response.setRemark(null);
                return response;
            }

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

首先调用`org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#pickupTopicRouteData`从`org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#topicQueueTable`获取到`org.apache.rocketmq.common.protocol.route.QueueData`这里面存了 brokerName，再通过`org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#brokerAddrTable`里获取到 broker 的地址信息等，然后再获取 orderMessage 的配置。

简要分析了下 RocketMQ 的 NameServer 的代码，比较粗粒度。

Related Posts

- 聊聊 RocketMQ 的 Broker 源码
- 聊一下 RocketMQ 的 DefaultMQPushConsumer 源码
- 介绍一下 RocketMQ

请我喝杯咖啡

WeChat Pay

Alipay

- Post author: Nicksxs
- Post link: https://nicksxs.me/2020/07/05/聊一下-RocketMQ-的-NameServer-源码/
- Copyright Notice: All articles in this blog are licensed under BY-NC-SA unless stating additionally.