标签：环境 json broker buffer remember sse 开头 into pack
说在前面
本次开始进行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();
说在最后
本次解析仅代表个人观点，仅供参考。
rocketmq源码解析之NamesrvController创建
标签：环境 json broker buffer remember sse 开头 into pack
原文地址：https://www.cnblogs.com/tianheframe/p/11615246.html
