RocketMQ源码解析一（NameServer启动流程）

秒速五厘米 2022-09-12 07:46 81阅读 0赞

RocketMQ版本4.6.0，记录自己看源码的过程

从启动类开始看

  1. public class NamesrvStartup { 
  3.     // 不懂，应该跟日志，配置和命令行参数有关
  4.     private static InternalLogger log;
  5.     private static Properties properties = null;
  6.     private static CommandLine commandLine = null;
  8.     public static void main(String[] args) { 
  9.         main0(args);
  10.     }
  12.     public static NamesrvController main0(String[] args) { 
  14.         try { 
  15.             // 解析配置并创建NamesrvController这一核心处理请求组件
  16.             NamesrvController controller = createNamesrvController(args);
  17.             // 初始化并启动NamesrvController，创建netty服务器，请求处理器，后台定时任务等
  18.             start(controller);
  19.             String tip = "The Name Server boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
  20.             log.info(tip);
  21.             System.out.printf("%s%n", tip);
  22.             return controller;
  23.         } catch (Throwable e) { 
  24.             e.printStackTrace();
  25.             System.exit(-1);
  26.         }
  28.         return null;
  29.     }
  30. }

首先第一步就是解析配置并创建NamesrvController这一核心处理请求组件

  1. public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException { 
  2.     System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
  3.     //PackageConflictDetect.detectFastjson();
  5.     Options options = ServerUtil.buildCommandlineOptions(new Options());
  6.     commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());
  7.     if (null == commandLine) { 
  8.         System.exit(-1);
  9.         return null;
  10.     }
  12.     // 创建两个核心配置，后面要根据这两个配置创建NamesrvController
  13.     final NamesrvConfig namesrvConfig = new NamesrvConfig();
  14.     final NettyServerConfig nettyServerConfig = new NettyServerConfig();
  15.     // netty监听9876端口
  16.     nettyServerConfig.setListenPort(9876);
  17.     if (commandLine.hasOption('c')) { 
  18.         String file = commandLine.getOptionValue('c');
  19.         if (file != null) { 
  20.             InputStream in = new BufferedInputStream(new FileInputStream(file));
  21.             properties = new Properties();
  22.             properties.load(in);
  23.             MixAll.properties2Object(properties, namesrvConfig);
  24.             MixAll.properties2Object(properties, nettyServerConfig);
  26.             namesrvConfig.setConfigStorePath(file);
  28.             System.out.printf("load config properties file OK, %s%n", file);
  29.             in.close();
  30.         }
  31.     }
  33.     if (commandLine.hasOption('p')) { 
  34.         InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
  35.         MixAll.printObjectProperties(console, namesrvConfig);
  36.         MixAll.printObjectProperties(console, nettyServerConfig);
  37.         System.exit(0);
  38.     }
  40.     MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);
  42.     if (null == namesrvConfig.getRocketmqHome()) { 
  43.         System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation%n", MixAll.ROCKETMQ_HOME_ENV);
  44.         System.exit(-2);
  45.     }
  47.     LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
  48.     JoranConfigurator configurator = new JoranConfigurator();
  49.     configurator.setContext(lc);
  50.     lc.reset();
  51.     configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");
  53.     log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
  55.     MixAll.printObjectProperties(log, namesrvConfig);
  56.     MixAll.printObjectProperties(log, nettyServerConfig);
  58.     // 根据上面的配置创建NamesrvController
  59.     final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
  61.     // remember all configs to prevent discard
  62.     controller.getConfiguration().registerConfig(properties);
  64.     return controller;
  65. }

前面几行看不懂直接跳过，直接看这3行关键的代码：

  1. // 创建两个核心配置，后面要根据这两个配置创建NamesrvController
  2. final NamesrvConfig namesrvConfig = new NamesrvConfig();
  3. final NettyServerConfig nettyServerConfig = new NettyServerConfig();
  4. // netty监听9876端口
  5. nettyServerConfig.setListenPort(9876);

这里应该可以猜测是为创建NamesrvController提供一些配置，看看NamesrvConfig和NettyServerConfig是什么

  1. public class NamesrvConfig { 
  2.     private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
  4.     // rocketmq的主目录，通过读取ROCKETMQ_HOME环境变量
  5.     private String rocketmqHome = System.getProperty(
  6.         MixAll.ROCKETMQ_HOME_PROPERTY, 
  7.         System.getenv(MixAll.ROCKETMQ_HOME_ENV));
  9.     // kv配置文件的路径
  10.     private String kvConfigPath = System.getProperty("user.home") 
  11.         + File.separator 
  12.         + "namesrv" 
  13.         + File.separator 
  14.         + "kvConfig.json";
  16.     // namesrv自己的配置文件路径
  17.     private String configStorePath = System.getProperty("user.home") 
  18.         + File.separator 
  19.         + "namesrv" 
  20.         + File.separator 
  21.         + "namesrv.properties";
  23.     private String productEnvName = "center";
  25.     private boolean clusterTest = false;
  27.     private boolean orderMessageEnable = false;
  28. }

这里主要配置一些路径，对于理解主线流程帮助不大，可以跳过。

  1. public class NettyServerConfig implements Cloneable { 
  2.     // 监听端口，但nameserver改为了9876
  3.     private int listenPort = 8888;
  4.     // 工作线程
  5.     private int serverWorkerThreads = 8;
  6.     // netty的public线程池的线程数量
  7.     private int serverCallbackExecutorThreads = 0;
  8.     // 这是netty的IO线程池的线程数量，默认是3，这里的线程是负责解析网络请求的
  9.     // 它这里的线程解析完网络请求之后，就会把请求转发给work线程来处理
  10.     private int serverSelectorThreads = 3;
  12.     private int serverOnewaySemaphoreValue = 256;
  13.     private int serverAsyncSemaphoreValue = 64;
  15.     // 如果一个网络连接空闲超过120s，就会被关闭
  16.     private int serverChannelMaxIdleTimeSeconds = 120;
  18.     private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;
  19.     private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;
  20.     private boolean serverPooledByteBufAllocatorEnable = true;
  22.     /** * make make install * * * ../glibc-2.10.1/configure \ --prefix=/usr \ --with-headers=/usr/include \ * --host=x86_64-linux-gnu \ --build=x86_64-pc-linux-gnu \ --without-gd */
  23.     private boolean useEpollNativeSelector = false;
  24. }

这是对NettyServer的一些配置信息，由于对netty不了解，所以这里也可以先跳过。

接下来的一些解析配置也不是很懂，就跳过了，直接看创建NamesrvController

  1. // 根据上面的配置创建NamesrvController
  2. final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);

可以看到是根据上面的两个核心配置创建的

  1. public NamesrvController(NamesrvConfig namesrvConfig, NettyServerConfig nettyServerConfig) { 
  2.     this.namesrvConfig = namesrvConfig;
  3.     this.nettyServerConfig = nettyServerConfig;
  4.     this.kvConfigManager = new KVConfigManager(this);
  5.     this.routeInfoManager = new RouteInfoManager();
  6.     this.brokerHousekeepingService = new BrokerHousekeepingService(this);
  7.     this.configuration = new Configuration(
  8.         log,
  9.         this.namesrvConfig, this.nettyServerConfig
  10.     );
  11.     this.configuration.setStorePathFromConfig(this.namesrvConfig, "configStorePath");
  12. }

创建的同时会对里面的一些管理器组件进行创建，供后续使用。

看入口方法，NameSrvController创建完成后就该进行初始化并启动，创建netty服务器，请求处理器，后台定时任务等。

  1. public static NamesrvController start(final NamesrvController controller) throws Exception { 
  3.     if (null == controller) { 
  4.         throw new IllegalArgumentException("NamesrvController is null");
  5.     }
  7.     // 进行初始化，创建netty服务器，请求处理器，后台定时任务等
  8.     boolean initResult = controller.initialize();
  9.     if (!initResult) { 
  10.         controller.shutdown();
  11.         System.exit(-3);
  12.     }
  14.     Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() { 
  15.         @Override
  16.         public Void call() throws Exception { 
  17.             controller.shutdown();
  18.             return null;
  19.         }
  20.     }));
  22.     // 其实就是启动netty服务器
  23.     controller.start();
  25.     return controller;
  26. }

这里面关键步骤就两步，初始化和启动，先看如何初始化NamesrvController：

  1. public boolean initialize() { 
  3.     // 无关紧要，跳过
  4.     this.kvConfigManager.load();
  6.     // 创建netty server
  7.     this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
  9.     // 默认请求处理器的线程池
  10.     this.remotingExecutor =
  11.         Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
  13.     // 注册默认请求处理器
  14.     this.registerProcessor();
  16.     // 每隔10s定时扫描broker状态的，看是否还存活着
  17.     this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  19.         @Override
  20.         public void run() { 
  21.             NamesrvController.this.routeInfoManager.scanNotActiveBroker();
  22.         }
  23.     }, 5, 10, TimeUnit.SECONDS);
  25.     // 忽略不重要代码
  27.     return true;
  28. }

这里面分为4步：
1、创建netty server，毕竟namesrv要作为服务端来接受broker，producer以及consumer的请求的，所以使用之前创建好的配置来创建netty server

  1. public NettyRemotingServer(final NettyServerConfig nettyServerConfig,
  2.     final ChannelEventListener channelEventListener) { 
  3.     super(nettyServerConfig.getServerOnewaySemaphoreValue(), nettyServerConfig.getServerAsyncSemaphoreValue());
  4.     this.serverBootstrap = new ServerBootstrap();
  5.     this.nettyServerConfig = nettyServerConfig;
  6.     this.channelEventListener = channelEventListener;
  7.     // 忽略看不懂的代码
  8. }

这里面最重要的一行代码就是this.serverBootstrap = new ServerBootstrap()。ServerBootstrap是netty里面的一个核心类，可以看作是netty的服务器，在监听9876端口。
2、创建一个默认请求处理器的线程池，用来处理没有专门处理器的请求，至于何谓专门，下面会介绍，但在namesrv中，就这一个处理器线程池，所以其实在namesrv，所有的请求都是由该线程池处理的。
3、注册默认请求处理器，跟上一步说明一样，namesrv只有这一个处理器。

  1. private void registerProcessor() { 
  2.     if (namesrvConfig.isClusterTest()) { 
  4.         this.remotingServer.registerDefaultProcessor(new ClusterTestRequestProcessor(this, namesrvConfig.getProductEnvName()),
  5.                                                      this.remotingExecutor);
  6.     } else { 
  7.         // 向netty服务器注册默认的请求处理组件
  8.         this.remotingServer.registerDefaultProcessor(new DefaultRequestProcessor(this), this.remotingExecutor);
  9.     }
  10. }

4、创建一个定时器，每隔10s定时扫描broker状态的，看是否还存活着，挂了就移除相关broker的信息

  1. /** * 扫描发生故障的broker，即默认120s没有更新心跳时间，则表示该broker挂了，需要将该broker路由信息移除 */
  2. public void scanNotActiveBroker() { 
  3.     // 迭代brokerLiveTable，判断每个broker的心跳时间跟当前时间的间隔
  4.     Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
  5.     while (it.hasNext()) { 
  6.         Entry<String, BrokerLiveInfo> next = it.next();
  7.         long last = next.getValue().getLastUpdateTimestamp();
  8.         // 默认超过120s就表示broker挂了
  9.         if ((last + BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) { 
  10.             RemotingUtil.closeChannel(next.getValue().getChannel());
  11.             it.remove();
  12.             log.warn("The broker channel expired, {} {}ms", next.getKey(), BROKER_CHANNEL_EXPIRED_TIME);
  13.             // 移除与该broker所有的信息，有点复杂，只知道移除所有有关broker信息就行了
  14.             this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
  15.         }
  16.     }
  17. }

这里面的一些实例变量等后面解析broker发送心跳时再具体说明。

最后就是启动NamesrvController这一组件：

  1. public void start() throws Exception { 
  2.     this.remotingServer.start();
  4.     if (this.fileWatchService != null) { 
  5.         this.fileWatchService.start();
  6.     }
  7. }

其实就是启动Netty服务器，至于启动细节，看不懂。

好了，到这里，namesrv服务就算是启动成功了，总的来说，namesrv的代码还是比较简单的。

总结一下，就是先解析相关配置，创建namesrvConfig和nettyServerConfig这两个核心配置，根据这两个配置创建NamesrvController这一 核心组件，创建出来以后对该组件进行初始化，包括创建netty服务器，请求处理器，请求线程池，后台定时任务等，最后启动NamesrvController，其实就是启动netty服务器。
在这里插入图片描述
下面就要开始研究NameServer启动成功后，Broker是如何启动的，如果向NameServer进行注册，如何进行心跳，NameServer是如何管理Broker的。

参考资料
《儒猿技术窝——从 0 开始带你成为消息中间件实战高手》

发表评论

表情：
评论列表 （有 0 条评论，81人围观）

还没有评论，来说两句吧...

相关阅读