RocketMq4.7源码解析之二(name,broker启动流程)

Love The Way You Lie 2022-11-30 04:06 277阅读 0赞

文章目录

  • broker启动流程
    • NameServer启动流程
      • mian
        • createNamesrvController
          • NamesrvController构造
      • start
        • initialize
        • start

在这里插入图片描述

broker启动流程

broker配置了配置文件地址，以及环境变量配置文件所在目录
在这里插入图片描述
vf

NameServer启动流程

mian

  1. public static NamesrvController main0(String[] args) {
  4.         try {
  6.             NamesrvController controller = createNamesrvController(args);
  7.             start(controller);
  8.             String tip = "The Name Server boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
  9.             log.info(tip);
  10.             System.out.printf("%s%n", tip);
  11.             return controller;
  12.         } catch (Throwable e) {
  14.             e.printStackTrace();
  15.             System.exit(-1);
  16.         }
  18.         return null;
  19.     }

createNamesrvController

  1. public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {
  3.         //设置版本号
  4.         System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
  5.         //PackageConflictDetect.detectFastjson();
  7.         Options options = ServerUtil.buildCommandlineOptions(new Options());
  8.         //解析命令行
  9.         commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());
  10.         if (null == commandLine) {
  11.     //命令行为null,则返回
  12.             System.exit(-1);
  13.             return null;
  14.         }
  16.         //names业务参数
  17.         //初次加载时,从环境变量中获取ROCKETMQ_HOME并赋值
  18.         final NamesrvConfig namesrvConfig = new NamesrvConfig();
  19.         //网络参数
  20.         final NettyServerConfig nettyServerConfig = new NettyServerConfig();
  21.         //设置监听端口
  22.         nettyServerConfig.setListenPort(9876);
  23.         //解析-c configFile 通过，c 命令指定配置文件的路径。
  24.         //使用“ 一属性名属性值”，例如一listenPort 9876 。
  25.         if (commandLine.hasOption('c')) {
  27.             String file = commandLine.getOptionValue('c');
  28.             if (file != null) {
  30.                 //读取配置文件
  31.                 InputStream in = new BufferedInputStream(new FileInputStream(file));
  32.                 properties = new Properties();
  33.                 properties.load(in);
  34.                 //解析配置文件属性
  35.                 MixAll.properties2Object(properties, namesrvConfig);
  36.                 MixAll.properties2Object(properties, nettyServerConfig);
  38.                 //设置配置文件路径
  39.                 namesrvConfig.setConfigStorePath(file);
  41.                 System.out.printf("load config properties file OK, %s%n", file);
  42.                 in.close();
  43.             }
  44.         }
  46.         //-p 打印当前加载配置属性
  47.         if (commandLine.hasOption('p')) {
  49.             InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
  50.             MixAll.printObjectProperties(console, namesrvConfig);
  51.             MixAll.printObjectProperties(console, nettyServerConfig);
  52.             System.exit(0);
  53.         }
  55.         //解析命令行属性填充到对象
  56.         MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);
  58.             //ROCKETMQ_HOME环境变量非空判断
  59.         if (null == namesrvConfig.getRocketmqHome()) {
  61.             System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation%n", MixAll.ROCKETMQ_HOME_ENV);
  62.             System.exit(-2);
  63.         }
  64.         //日志相关
  65.         LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
  66.         JoranConfigurator configurator = new JoranConfigurator();
  67.         configurator.setContext(lc);
  68.         lc.reset();
  69.         configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");
  71.         log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
  73.         //log打印name和netty配置属性
  74.         MixAll.printObjectProperties(log, namesrvConfig);
  75.         MixAll.printObjectProperties(log, nettyServerConfig);
  77.         final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
  79.         // remember all configs to prevent discard
  80.         //使用配置文件属性,注册并覆盖默认属性
  81.         controller.getConfiguration().registerConfig(properties);
  83.         return controller;
  84.     }
NamesrvController构造
  1. public NamesrvController(NamesrvConfig namesrvConfig, NettyServerConfig nettyServerConfig) {
  3.         this.namesrvConfig = namesrvConfig;
  4.         this.nettyServerConfig = nettyServerConfig;
  5.         this.kvConfigManager = new KVConfigManager(this);
  6.         //路由信息管理,内部有topic,broker,集群,以及心跳之类的信息
  7.         this.routeInfoManager = new RouteInfoManager();
  8.         this.brokerHousekeepingService = new BrokerHousekeepingService(this);
  9.         //合并对象成员变量属性,并保存对象
  10.         this.configuration = new Configuration(
  11.             log,
  12.             this.namesrvConfig, this.nettyServerConfig
  13.         );
  14.         this.configuration.setStorePathFromConfig(this.namesrvConfig, "configStorePath");
  15.     }
  17. public Configuration(InternalLogger log, Object... configObjects) {
  19.         this.log = log;
  20.         if (configObjects == null || configObjects.length == 0) {
  22.             return;
  23.         }
  24.             //合并对象成员变量属性,并保存对象
  25.         for (Object configObject : configObjects) {
  27.             registerConfig(configObject);
  28.         }
  29.     }
  30.     public Configuration registerConfig(Object configObject) {
  32.         try {
  34.             readWriteLock.writeLock().lockInterruptibly();
  35.             try {
  37.                 //对象成员变量值填充到prop
  38.                 Properties registerProps = MixAll.object2Properties(configObject);
  39.                 //属性覆盖allConfigs
  40.                 merge(registerProps, this.allConfigs);
  41.                 //记录对象
  42.                 configObjectList.add(configObject);
  43.             } finally {
  45.                 readWriteLock.writeLock().unlock();
  46.             }
  47.         } catch (InterruptedException e) {
  49.             log.error("registerConfig lock error");
  50.         }
  51.         return this;
  52.     }

start

  1. public static NamesrvController start(final NamesrvController controller) throws Exception {
  4.         if (null == controller) {
  6.             throw new IllegalArgumentException("NamesrvController is null");
  7.         }
  8.         //初始化controller
  9.         boolean initResult = controller.initialize();
  10.         //初始化失败,退出
  11.         if (!initResult) {
  13.             controller.shutdown();
  14.             System.exit(-3);
  15.         }
  16.         //注册JVM 钩子函数并启动服务器，
  17.         //在JVM 进程关闭之前，先将线程池关闭，及时释放资源。 tomcat也有使用这种方法释放资源
  18.         Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {
  20.             @Override
  21.             public Void call() throws Exception {
  23.                 controller.shutdown();
  24.                 return null;
  25.             }
  26.         }));
  28.         //启动
  29.         controller.start();
  31.         return controller;
  32.     }

initialize

  1. public boolean initialize() {
  3.         //加载KV 配置
  4.         this.kvConfigManager.load();
  5.         //创建NettyServer 网络处理对象,创建线程池和信号量
  6.         this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
  8.         //创建Netty业务线程池
  9.         this.remotingExecutor =
  10.             Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
  12.         this.registerProcessor();
  13.         //NameServer 每隔10s扫描一次Broker ， 移除处于不激活状态的Broker
  14.         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
  17.             @Override
  18.             public void run() {
  20.                 NamesrvController.this.routeInfoManager.scanNotActiveBroker();
  21.             }
  22.         }, 5, 10, TimeUnit.SECONDS);
  23.         //NameServer每隔10 分钟打印一次KV 配置。
  24.         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
  27.             @Override
  28.             public void run() {
  30.                 NamesrvController.this.kvConfigManager.printAllPeriodically();
  31.             }
  32.         }, 1, 10, TimeUnit.MINUTES);
  34.         if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {
  36.             // Register a listener to reload SslContext
  37.             try {
  39.                 fileWatchService = new FileWatchService(
  40.                     new String[] {
  42.                         TlsSystemConfig.tlsServerCertPath,
  43.                         TlsSystemConfig.tlsServerKeyPath,
  44.                         TlsSystemConfig.tlsServerTrustCertPath
  45.                     },
  46.                     new FileWatchService.Listener() {
  48.                         boolean certChanged, keyChanged = false;
  49.                         @Override
  50.                         public void onChanged(String path) {
  52.                             if (path.equals(TlsSystemConfig.tlsServerTrustCertPath)) {
  54.                                 log.info("The trust certificate changed, reload the ssl context");
  55.                                 reloadServerSslContext();
  56.                             }
  57.                             if (path.equals(TlsSystemConfig.tlsServerCertPath)) {
  59.                                 certChanged = true;
  60.                             }
  61.                             if (path.equals(TlsSystemConfig.tlsServerKeyPath)) {
  63.                                 keyChanged = true;
  64.                             }
  65.                             if (certChanged && keyChanged) {
  67.                                 log.info("The certificate and private key changed, reload the ssl context");
  68.                                 certChanged = keyChanged = false;
  69.                                 reloadServerSslContext();
  70.                             }
  71.                         }
  72.                         private void reloadServerSslContext() {
  74.                             ((NettyRemotingServer) remotingServer).loadSslContext();
  75.                         }
  76.                     });
  77.             } catch (Exception e) {
  79.                 log.warn("FileWatchService created error, can't load the certificate dynamically");
  80.             }
  81.         }
  83.         return true;
  84.     }

start

  1. public void start() {
  3.         //Worker线程池
  4.         this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
  5.             nettyServerConfig.getServerWorkerThreads(),
  6.             new ThreadFactory() {
  9.                 private AtomicInteger threadIndex = new AtomicInteger(0);
  11.                 @Override
  12.                 public Thread newThread(Runnable r) {
  14.                     return new Thread(r, "NettyServerCodecThread_" + this.threadIndex.incrementAndGet());
  15.                 }
  16.             });
  18.         //初始化handler和编码器
  19.         prepareSharableHandlers();
  21.         ServerBootstrap childHandler =
  22.                 //设置EventLoopGroup,来接受和处理新的连接,以进行事件的处理，如接受新连接以及读/写数据
  23.                 //server socket接收了新建连接后，会把connection socket放到workerGroup中进行处理
  24.             this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
  25.                     //epoll还是nio channel
  26.                 .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
  27.                 .option(ChannelOption.SO_BACKLOG, 1024)
  28.                 .option(ChannelOption.SO_REUSEADDR, true)
  29.                 .option(ChannelOption.SO_KEEPALIVE, false)
  30.                 .childOption(ChannelOption.TCP_NODELAY, true)
  31.                 .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())
  32.                 .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())
  33.                     //使用指定的端口设置套接字地址
  34.                 .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))
  35.                 .childHandler(new ChannelInitializer<SocketChannel>() {
  37.                     @Override
  38.                     public void initChannel(SocketChannel ch) throws Exception {
  40.                         //分别添加2个handler,处理请求
  41.                         ch.pipeline()
  42.                             .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME, handshakeHandler)
  43.                             .addLast(defaultEventExecutorGroup,
  44.                                     //设置编码器
  45.                                 encoder,
  46.                                     //设置解码器
  47.                                 new NettyDecoder(),
  48.                                     /*
  49.                                     当连接空闲时间太长时，将会触发一个IdleStateEvent 事件.
  50.                                     ChannelInboundHandler中重写userEvent-Triggered()方法来处理该IdleStateEvent 事件
  51.                                      */
  52.                                 new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
  53.                                 connectionManageHandler,
  54.                                 serverHandler
  55.                             );
  56.                     }
  57.                 });
  59.         if (nettyServerConfig.isServerPooledByteBufAllocatorEnable()) {
  61.             childHandler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
  62.         }
  64.         try {
  66.             //异步地绑定服务器；调用 sync()方法阻塞等待直到绑定完成
  67.             ChannelFuture sync = this.serverBootstrap.bind().sync();
  68.             //记录port
  69.             InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
  70.             this.port = addr.getPort();
  71.         } catch (InterruptedException e1) {
  73.             throw new RuntimeException("this.serverBootstrap.bind().sync() InterruptedException", e1);
  74.         }
  76.         if (this.channelEventListener != null) {
  78.             this.nettyEventExecutor.start();
  79.         }
  81.         this.timer.scheduleAtFixedRate(new TimerTask() {
  84.             @Override
  85.             public void run() {
  87.                 try {
  89.                     NettyRemotingServer.this.scanResponseTable();
  90.                 } catch (Throwable e) {
  92.                     log.error("scanResponseTable exception", e);
  93.                 }
  94.             }
  95.         }, 1000 * 3, 1000);
  96.     }

发表评论

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

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

相关阅读