RocketMQ中的NameServer详细解析

public static NamesrvController main0(String[] args) {
//构造NamesrvController 
            NamesrvController controller = createNamesrvController(args);
            start(controller);
            return controller;
}
    public static NamesrvController start(final NamesrvController controller) throws Exception {
//初始化
        boolean initResult = controller.initialize();
//启动
        controller.start();
        return controller;
    }
 

public class NamesrvController {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
 
    //nameSrv的配置
    private final NamesrvConfig namesrvConfig;
    //netty的配置
    private final NettyServerConfig nettyServerConfig;
 
    //执行单线程的任务调度，自定义编程名称
    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl(
        "NSScheduledThread"));
 
    //kv的配置管理
    private final KVConfigManager kvConfigManager;
 
    //路由管理器，有broker的ip和队列信息，producer发送的queue信息，consumer的pull的queue信息
    private final RouteInfoManager routeInfoManager;
 
    //namesrv的netty的服务端实现
    private RemotingServer remotingServer;
 
    //处理接受到请求事件的回调监听服务，主要处理netty的事件
    private BrokerHousekeepingService brokerHousekeepingService;
 
    private ExecutorService remotingExecutor;
 
    private Configuration configuration;
    private FileWatchService fileWatchService;
 
    public NamesrvController(NamesrvConfig namesrvConfig, NettyServerConfig nettyServerConfig) {
        this.namesrvConfig = namesrvConfig;
        this.nettyServerConfig = nettyServerConfig;
        //构造kv配置的管理
        this.kvConfigManager = new KVConfigManager(this);
        //构造路由信息管理
        this.routeInfoManager = new RouteInfoManager();
        //构造网络连接事件管理
        this.brokerHousekeepingService = new BrokerHousekeepingService(this);
        //配置
        this.configuration = new Configuration(
            log,
            this.namesrvConfig, this.nettyServerConfig
        );
        this.configuration.setStorePathFromConfig(this.namesrvConfig, "configStorePath");
    }

    public boolean initialize() {
 
        //加载kv配置
        this.kvConfigManager.load();
        //初始化NettyServer
        this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
        //执行器，用于接受请求并进行处理
        this.remotingExecutor =
            Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
        //注册Netty的处理器，即remotingExecutor
        this.registerProcessor();
        //进行扫描未活跃的Broker的任务
        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);
        //...省略其他代码
        return true;
    }
 
   
    private void registerProcessor() {
        if (namesrvConfig.isClusterTest()) {
            this.remotingServer.registerDefaultProcessor(new ClusterTestRequestProcessor(this, namesrvConfig.getProductEnvName()),
                this.remotingExecutor);
        } else {
            //注册request处理器
            this.remotingServer.registerDefaultProcessor(new DefaultRequestProcessor(this), this.remotingExecutor);
        }
    }  
  
 
    public void start() throws Exception {
        //启动netty server
        this.remotingServer.start();
 
        if (this.fileWatchService != null) {
            this.fileWatchService.start();
        }
    }

@Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        //...
        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);
                }
            //...省略其他代码
        }
        return null;
    }