RocketMQ之NameServer架构设计及启动关闭流程源码分析

目录

• NameServer

• 1.架构设计
• 2.核心类与配置

• NamesrvController
• NamesrvConfig
• NettyServerConfig
• RouteInfoManager

• 3.1.步骤一

private final NamesrvConfig namesrvConfig;//主要指定 nameserver 的相关配置属性
private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("NSScheduledThread"));//NameServer定时任务执行线程池-->每隔10s扫描broker，对存活的Broker信息进行维护并且打印KVConfig
private final KVConfigManager kvConfigManager;//读取或变更NameServer的配置属性，加载 NamesrvConfig中配置到内存
private final RouteInfoManager routeInfoManager;//NameServer 数据的载体，记录 Broker、Topic 等信息。
private final NettyServerConfig nettyServerConfig;//与网络通讯相关的配置
private RemotingServer remotingServer;//网络通信服务
private ExecutorService remotingExecutor;//网络通信服务

private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
private String kvConfigPath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "kvConfig.json";
private String configStorePath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "namesrv.properties";
private String productEnvName = "center";
private boolean clusterTest = false;
private boolean orderMessageEnable = false;

private int listenPort = 8888;
private int serverWorkerThreads = 8;
private int serverCallbackExecutorThreads = 0;
private int serverSelectorThreads = 3;
private int serverOnewaySemaphoreValue = 256;
private int serverAsyncSemaphoreValue = 64;
private int serverChannelMaxIdleTimeSeconds = 120;
private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;
private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;
private boolean serverPooledByteBufAllocatorEnable = true;

private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
private final ReadWriteLock lock = new ReentrantReadWriteLock();
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;

private String brokerName;			//broker名称
private int readQueueNums;			//读队列个数
private int writeQueueNums;			//写队列个数
private int perm;				   //操作权限	
private int topicSysFlag;			//同步复制还是异步复制的标识

private String cluster;				//当前Broker所属集群
private String brokerName;			//Broker名称
//BrokerId=0表示主节点，BrokerId>0表示从节点
//记录BrokerId与对应节点地址的映射信息
private HashMap<Long/* brokerId */, String/* broker address */> brokerAddrs;

private long lastUpdateTimestamp;		//上次发送心跳包的时间戳
private DataVersion dataVersion;		//记录数据版本信息
private Channel channel;
private String haServerAddr;			//Master节点地址

public static void main(String[] args) {
    main0(args);
}
public static NamesrvController main0(String[] args) {
    try {
        //创建NamesrvController的入口
        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;
}

public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {
    //....    
    //创建namesrvConfig
    final NamesrvConfig namesrvConfig = new NamesrvConfig();
    //创建nettyServerConfig
    final NettyServerConfig nettyServerConfig = new NettyServerConfig();
    //设置默认端口9876
    nettyServerConfig.setListenPort(9876);
    //-c 指定属性配置文件的位置
    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 属性名=属性值
    if (commandLine.hasOption('p')) {
        InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
        MixAll.printObjectProperties(console, namesrvConfig);
        MixAll.printObjectProperties(console, nettyServerConfig);
        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);
    }
	//....
    //打印配置信息日志
    MixAll.printObjectProperties(log, namesrvConfig);
    MixAll.printObjectProperties(log, nettyServerConfig);
    //根据namesrvConfig和nettyServerConfig创建NamesrvController
    final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
    // 将配置存入controller.configuration以防止配置丢失
    controller.getConfiguration().registerConfig(properties);
    return controller;
}

//NamesrvStartup#start
public static NamesrvController start(final NamesrvController controller) throws Exception {
    //进行简单的检查
    if (null == controller) {
        throw new IllegalArgumentException("NamesrvController is null");
    }	
    //controller初始化
    boolean initResult = controller.initialize();
    if (!initResult) {
        controller.shutdown();
        System.exit(-3);
    }    
    //....
    controller.start();		//开启远程服务-this.remotingServer.start();
    return controller;
}
//NamesrvController#initialize
public boolean initialize() {
    	//加载配置管理器
        this.kvConfigManager.load();
		//创建Netty远程服务
        this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
		//创建远程服务线程池
        this.remotingExecutor =
            Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
        //注册线程池
        this.registerProcessor();
        //定时任务线程池--->每隔十秒扫描活跃状态异常的Broker信息
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            /**
             * 对Not Active Broker 进行扫描
             */
            @Override
            public void run() {
                NamesrvController.this.routeInfoManager.scanNotActiveBroker();
            }
        }, 5, 10, TimeUnit.SECONDS);
        //定时任务线程池--->每隔十秒打印KVConfig信息
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                NamesrvController.this.kvConfigManager.printAllPeriodically();
            }
        }, 1, 10, TimeUnit.MINUTES);
       //....
        return true;
    }

public static NamesrvController start(final NamesrvController controller) throws Exception {
	//....    
    //JVM进程关闭之前,将线程池关闭,资源释放
    Runtime.getRuntime().addShutdownHook/*注册JVM钩子函数*/(new ShutdownHookThread(log, new Callable<Void>() {
        @Override
        public Void call() throws Exception {
            controller.shutdown();
            return null;
        }
    }));	
    //....
}