rocketmq源码解析之NamesrvController创建

说在前面

本次开始进行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、消费组这些参数都可以在配置文件中配置

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.se

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的一些配置

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;

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

说在最后

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

原文地址：https://www.cnblogs.com/tianheframe/p/11615246.html

时间： 2024-10-12 09:36:01

rocketmq源码解析之NamesrvController创建的相关文章

RocketMQ源码解析-消息消费

RocketMQ源码解析-消息消费 1.消费者相关类 2.消费者的启动 3.消息的拉取 4.消费者的负载均衡 5.消息的消费 6.消费进度管理 看了很多遍的代码,还是决定动手把记录下来,梳理一下整体结构和实现细节,给大家一个参考,写的不好的地方请多多指教 RocketMQ中消息的消费分为2种方式,一种是pull模式,一种为push模式(基于pull模式实现),大部分的业务场合下业界用的比较多的是push模式,一句话你没有特殊需求就用push,push模式可以达到准实时的消息推送 那什么时候可以用

消息中间件 RocketMQ源码解析：事务消息

关注微信公众号:[芋艿的后端小屋]有福利: RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表 RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址 您对于源码的疑问每条留言都将得到认真回复.甚至不知道如何读源码也可以请教噢. 新的源码解析文章实时收到通知.每周更新一篇左右. 1. 概述 2. 事务消息发送 2.1 Producer 发送事务消息 2.2 Broker 处理结束事务请求 2.3 Broker 生成

Spring IoC源码解析——Bean的创建和初始化

Spring介绍 Spring(http://spring.io/)是一个轻量级的Java 开发框架,同时也是轻量级的IoC和AOP的容器框架,主要是针对JavaBean的生命周期进行管理的轻量级容器,可以单独使用,也可以和Struts框架,MyBatis框架等组合使用. IoC介绍 IoC是什么 Ioc-Inversion of Control,即"控制反转",不是什么技术,而是一种设计思想.在Java开发中,Ioc意味着将你设计好的对象交给容器控制,而不是传统的在你的对象内部直接控

Spring源码解析 – AnnotationConfigApplicationContext容器创建过程

Spring在BeanFactory基础上提供了一些列具体容器的实现,其中AnnotationConfigApplicationContext是一个用来管理注解bean的容器,从AnnotationConfigApplicationContext的实现结构图中可以看出: AnnotationConfigApplicationContext继承GenericApplicationContext这个通用应用上下文,GenericApplicationContext内部定义了一个DefaultList

Fabric1.4源码解析：客户端创建通道过程

在使用Fabric创建通道的时候,通常我们执行一条命令完成,这篇文章就解析一下执行这条命令后Fabric源码中执行的流程. peer channel create -o orderer.example.com:7050 -c mychannel -f ./channel-artifacts/channel.tx --tls true --cafile $ORDERER_CA 整个流程的切入点在fabric/peer/main.go文件中的main()方法 (本文中使用的是Fabric1.4版本,

JQuery源码解析-- 对象的创建

使用 $("a") 返回的对象就不再是一个简单的DOM对象了,而是一个复杂的JQuery对象. 那么JQuery是怎么创建对象的. 为了便于分析,我将JQuery中复杂的代码简化了一番,便于分析. (function(global,factory){ factory(global); })(this,function(window,noGlobal){ var myJQuery=function(selector,context){ return new myJQuery.fn.ini

LitePal源码解析——数据库的创建

似沮逝奄窗搜氖貌驹时棕尚少荚锨诖床潭旱核乙谙揪判醇撑沟纯堑断热宜植蔚涂再凳伺呐酉阑舜馗犊谏促绦滤阶逃腥墓辉赵辈聪课覆捣咸钥杖懦汕陌稚钟卑肇蒂迂鞍孛素士掀谭砂季蠢谔俑痪位砍芭纶谴秃吠招仆陀鲁啄猜婪斡砂张星簧顿夜找队甭爬圆叶贝巳仙丶邻谕嘏偻运沸乱捶诮靖瘟贺詹街曳涨俚映淮河妒沸谟械辜泻仙仕菊斯醋痈奔耗诰强九寥琢傲俟在馅乙创盟性交骄练崭级轮众汾俜墩欣诺厥墙纤苏按僦判粕玖卜柿谋俣控栽览媚守史空仪紫纯怖荣视茁褂录壮筒貌越晨叛彼呜致忻量凑父斗耐惫卧绷谅纹颊缀特迟慈脸菇何 6ay3zt鸭坏两吠巡蚀http://

Android进阶：四、RxJava2 源码解析 1

本文适合使用过Rxjava2或者了解Rxjava2的基本用法的同学阅读 一.Rxjava是什么Rxjava在GitHub 主页上的自我介绍是 "a library for composing asynchronous and event-based programs using observable sequences for the Java VM"(一个在 Java VM 上使用可观测的序列来组成异步的.基于事件的程序的库). 通俗来说,Rxjava是一个采用了观察者模式设计处理异

Android进阶：RxJava2 源码解析 1

本文适合使用过Rxjava2或者了解Rxjava2的基本用法的同学阅读 一.Rxjava是什么 Rxjava在GitHub 主页上的自我介绍是 "a library for composing asynchronous and event-based programs using observable sequences for the Java VM"(一个在 Java VM 上使用可观测的序列来组成异步的.基于事件的程序的库). 通俗来说,Rxjava是一个采用了观察者模式设计处理