rocketmq源码解析之NamesrvController创建

不念不忘少年蓝@ 2022-01-13 13:17 283阅读 0赞

[2019独角兽企业重金招聘Python工程师标准>>> ][2019_Python_] ![hot3.png][]

**说在前面**

本次开始进行rocketmq源码解析，比较喜欢rocketmq的架构设计，rocketmq内嵌了namesrv注册中心保存了元数据，进行负载均衡、容错的一些处理，4.3以上支持消息事务，有管理控制台、命令行工具，底层namesrv与broker、client与server交互netty实现。更多精彩文章请关注“天河聊架构”微信公众号。

**源码解析**

创建NamesrvController，进入这个方法org.apache.rocketmq.namesrv.NamesrvStartup\#main，再进入这个方法org.apache.rocketmq.namesrv.NamesrvStartup\#main0

public static NamesrvController main0(String[] args) {
    
            try {
    //            源码解析之创建namesrv控制器 =》
                NamesrvController controller = createNamesrvController(args);
    //            源码解析之启动namesrv控制器 =》
                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;
        }

进入这个方法org.apache.rocketmq.namesrv.NamesrvStartup\#createNamesrvController

public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {
    //        从系统文件中查询rocketmq版本，默认4.3.0
            System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
    //        PackageConflictDetect.detectFastjson();
    //        构建命令行操作的指令 =》
            Options options = ServerUtil.buildCommandlineOptions(new Options());
    //        mqnamesrv 启动namesrv命令 =》
            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();
    //        设置 namesrv的服务端口
            nettyServerConfig.setListenPort(9876);
    //        c 指定启动的时候加载配置文件
            if (commandLine.hasOption('c')) {
    //            命令行启动指定配置文件，前面用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);
    //                设置命令行启动namesrv指定的配置文件路径
                    namesrvConfig.setConfigStorePath(file);
                    System.out.printf("load config properties file OK, %s%n", file);
                    in.close();
                }
            }
    
    //        打印namesrv的配置信息，命令行上面加p
            if (commandLine.hasOption('p')) {
                MixAll.printObjectProperties(null, namesrvConfig);
                MixAll.printObjectProperties(null, nettyServerConfig);
    //            正常程序退出
                System.exit(0);
            }
    
    //        把命令行属性解析成properties
            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);
            }
    
            LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
            JoranConfigurator configurator = new JoranConfigurator();
            configurator.setContext(lc);
            lc.reset();
    //        logback日志文件路径
            configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");
            log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
            MixAll.printObjectProperties(log, namesrvConfig);
            MixAll.printObjectProperties(log, nettyServerConfig);
    //        创建namesrv控制器 =》
            final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
            // remember all configs to prevent discard 把配置文件配置值的属性值注册到namesrv控制器
            controller.getConfiguration().registerConfig(properties);
            return controller;
        }

进入这个方法org.apache.rocketmq.srvutil.ServerUtil\#buildCommandlineOptions 这里可以看到启动rocketmq的时候namesrv参数怎么制定

public static Options buildCommandlineOptions(final Options options) {
            Option opt = new Option("h", "help", false, "Print help");
            opt.setRequired(false);
            options.addOption(opt);
    //        这里可以看到启动的时候有一个n参数指定namesrv的地址，可以是单机可以是集群，启动地址之间用;分开
            opt =
                new Option("n", "namesrvAddr", true,
                    "Name server address list, eg: 192.168.0.1:9876;192.168.0.2:9876");
            opt.setRequired(false);
            options.addOption(opt);
            return options;
        }

进入这个方法org.apache.rocketmq.namesrv.NamesrvStartup\#buildCommandlineOptions 从这里看出来启动rocketmq的时候怎么指定配置文件，是否自动创建topic、消费组这些参数都可以在配置文件中配置

//    指定namesrv启动的时候加载的配置文件
        public static Options buildCommandlineOptions(final Options options) {
            Option opt = new Option("c", "configFile", true, "Name server config properties file");
            opt.setRequired(false);
            options.addOption(opt);
    //        控制台输出配置项
            opt = new Option("p", "printConfigItem", false, "Print all config item");
            opt.setRequired(false);
            options.addOption(opt);
            return options;
        }

从这里可以看到一些配置的存储地址

public class NamesrvConfig {
        private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    //    解析ROCKETMQ_HOME
        private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
    //    kvConfig存储地址
        private String kvConfigPath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "kvConfig.json";
    //    namesrv存储地址
        private String configStorePath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "namesrv.properties";

nettyserver的一些配置

public class NettyServerConfig implements Cloneable {
    //    netty server监听端口
        private int listenPort = 8888;
    //    服务端工作线程数
        private int serverWorkerThreads = 8;
    //    服务端回调执行线程
        private int serverCallbackExecutorThreads = 0;
    //    服务端选择器线程
        private int serverSelectorThreads = 3;
    //    oneway方式信号量的值
        private int serverOnewaySemaphoreValue = 256;
    //    服务端异步信号量值
        private int serverAsyncSemaphoreValue = 64;
    //    服务端channel最大空闲时间
        private int serverChannelMaxIdleTimeSeconds = 120;
    //    发送消息最大大小从系统属性中获取，默认值65535
        private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;
    //    接受消息最大大小从系统属性中获取，默认值65535
        private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;
        private boolean serverPooledByteBufAllocatorEnable = true;

nettyserver默认监听端口9876

nettyServerConfig.setListenPort(9876);

从这里可以看到是解析启动的时候指定的配置文件属性

//        c 指定启动的时候加载配置文件
            if (commandLine.hasOption('c')) {
    //            命令行启动指定配置文件，前面用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);

配置文件中可以配置这个类org.apache.rocketmq.common.namesrv.NamesrvConfig namesrv的一些配置

//    解析ROCKETMQ_HOME
        private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
    //    kvConfig存储地址
        private String kvConfigPath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "kvConfig.json";
    //    namesrv存储地址
        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;

也可以配置这个类org.apache.rocketmq.remoting.netty.NettyServerConfig nettyserver的一些配置

namesrv配置的存储地址就是启动指定的配置文件

namesrvConfig.setConfigStorePath(file);

也可以在命令行上指定namesrv属性配置

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

logback的配置文件

configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");

根据namesrv配置和nettyserver配置创建NamesrvController

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

进入这个方法org.apache.rocketmq.namesrv.NamesrvController\#NamesrvController

public NamesrvController(NamesrvConfig namesrvConfig, NettyServerConfig nettyServerConfig) {
            this.namesrvConfig = namesrvConfig;
            this.nettyServerConfig = nettyServerConfig;
            this.kvConfigManager = new KVConfigManager(this);
            this.routeInfoManager = new RouteInfoManager();
            this.brokerHousekeepingService = new BrokerHousekeepingService(this);
    //        =》
            this.configuration = new Configuration(
                log,
                this.namesrvConfig, this.nettyServerConfig
            );
    //        指定保存namesrv配置的配置文件
            this.configuration.setStorePathFromConfig(this.namesrvConfig, "configStorePath");
        }

进入这里org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager\#RouteInfoManager

public RouteInfoManager() {
    //        指定长度，减少扩容提高性能
            this.topicQueueTable = new HashMap<String, List<QueueData>>(1024);//topicQueue缓存信息
            this.brokerAddrTable = new HashMap<String, BrokerData>(128);//broker地址缓存
            this.clusterAddrTable = new HashMap<String, Set<String>>(32);//集群地址缓存
            this.brokerLiveTable = new HashMap<String, BrokerLiveInfo>(256);//可用的broker缓存
            this.filterServerTable = new HashMap<String, List<String>>(256);//过滤的server地址
        }

设置namrsrv配置存储地址System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "namesrv.properties"

this.configuration.setStorePathFromConfig(this.namesrvConfig, "configStorePath");

// remember all configs to prevent discard 把配置文件配置值的属性值注册到namesrv控制器
    controller.getConfiguration().registerConfig(properties);

进入这个方法org.apache.rocketmq.common.Configuration\#registerConfig(java.util.Properties)

public Configuration registerConfig(Object configObject) {
        try {
            readWriteLock.writeLock().lockInterruptibly();
            try {
    
                Properties registerProps = MixAll.object2Properties(configObject);
                merge(registerProps, this.allConfigs);
                configObjectList.add(configObject);
            } finally {
                readWriteLock.writeLock().unlock();
            }
        } catch (InterruptedException e) {
            log.error("registerConfig lock error");
        }
        return this;
    }

/**
     * All properties include configs in object and extend properties.
     * 所有的启动配置都在这里
     */
    private Properties allConfigs = new Properties();

**说在最后**

本次解析仅代表个人观点，仅供参考。

加入技术微信群

![84feb4fddc3bcfcc99c771803399197eacb.jpg][]

转载于:https://my.oschina.net/u/3775437/blog/3036175

[2019_Python_]: https://my.oschina.net/u/2663968/blog/3061697
[hot3.png]: /images/20220113/22ae98f2a8b140c8a2a7ff4dd6af4672.png
[84feb4fddc3bcfcc99c771803399197eacb.jpg]: /images/20220113/37bc2228a64248b8a69d4fa1b3f347f4.png