RocketMQ源码分析【二】NameServer启动流程源码分析

落日映苍穹つ 2022-09-12 04:54 403阅读 0赞

文章目录

    • 入口
    • NamesrvController是如何被创建出来的
    • 启动NameServer
      • 定时任务线程池
      • 构造Netty相关线程池
      • 启动Netty服务器
    • 总结

入口

NameServer 源码的启动入口在类NamesrvStartup中的main方法中
在这里插入图片描述
可以看到方法封装的非常好，每个方法干一件事。核心方法就是这两个

  1. NamesrvController controller = createNamesrvController(args);
  2.     start(controller);

NamesrvController是如何被创建出来的

我们看看createNamesrvController()方法具体干了什么

  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.         final NamesrvConfig namesrvConfig = new NamesrvConfig();
  13.         final NettyServerConfig nettyServerConfig = new NettyServerConfig();
  14.         nettyServerConfig.setListenPort(9876);
  15.         // 启动namesrv 参数带c代表读取配置文件
  16.         if (commandLine.hasOption('c')) { 
  17.             String file = commandLine.getOptionValue('c');
  18.             if (file != null) { 
  19.                 // 读取文件
  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.         // 如果启动参数带p 打印NameSrv启动参数
  34.         if (commandLine.hasOption('p')) { 
  35.             InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
  36.             MixAll.printObjectProperties(console, namesrvConfig);
  37.             MixAll.printObjectProperties(console, nettyServerConfig);
  38.             System.exit(0);
  39.         }
  40.         // 获取配置类容填充到namesrvConfig
  41.         MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);
  42.         // 如果rocketmqHome 打印异常退出
  43.         if (null == namesrvConfig.getRocketmqHome()) { 
  44.             System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation%n", MixAll.ROCKETMQ_HOME_ENV);
  45.             System.exit(-2);
  46.         }
  47.         // 日志相关配置
  48.         LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
  49.         JoranConfigurator configurator = new JoranConfigurator();
  50.         configurator.setContext(lc);
  51.         lc.reset();
  52.         configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");
  53.         // 打印NameServer所有配置信息
  54.         log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
  56.         MixAll.printObjectProperties(log, namesrvConfig);
  57.         MixAll.printObjectProperties(log, nettyServerConfig);
  59.         final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
  61.         // remember all configs to prevent discard
  62.         controller.getConfiguration().registerConfig(properties);
  64.         return controller;
  65.     }

上面源码比较长，都是解析参数构造一些配置类，这里作一个总结
在这里插入图片描述
这里构造的配置类我们没看下面的代码也不知道是干嘛的，所以我们先暂时不作过多分析，只用知道是解析了启动脚本的配置参数或设置的配置文件然后生成了对应的配置类

启动NameServer

启动NameServer的核心入口方法是

  1. start(controller);

我们看看这段代码的实现

  1. public static NamesrvController start(final NamesrvController controller) throws Exception { 
  3.         if (null == controller) { 
  4.             throw new IllegalArgumentException("NamesrvController is null");
  5.         }
  6.         // 初始化核心逻辑
  7.         boolean initResult = controller.initialize();
  8.         if (!initResult) { 
  9.             controller.shutdown();
  10.             System.exit(-3);
  11.         }
  13.         // 注册钩子函数 在JVM进程关闭时，优雅地释放netty服务、线程池等资源
  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.         }));
  21.         controller.start();
  22.         return controller;
  23.     }

可以看到这段方法中也没有什么核心实现。只是注册了一个钩子函数，在JVM关闭的时候去释放一些线程池、Netty服务器资源等。核心实现还是封装在方法

  1. boolean initResult = controller.initialize();

中，我们来重点看看这个方法的实现

  1. public boolean initialize() { 
  3.         this.kvConfigManager.load();
  4.         // 构建Netty服务器
  5.         this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
  6.         // netty 工作线程
  7.         this.remotingExecutor =           Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
  9.         // 将工作线程池给netty
  10.         this.registerProcessor();
  11.         // 定时检测不活跃的Broker
  12.         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  13.             @Override
  14.             public void run() { 
  15.                 NamesrvController.this.routeInfoManager.scanNotActiveBroker();
  16.             }
  17.         }, 5, 10, TimeUnit.SECONDS);
  19.         // 创建定时任务--每个10分钟打印一遍KV配置
  20.         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  21.             @Override
  22.             public void run() { 
  23.                 NamesrvController.this.kvConfigManager.printAllPeriodically();
  24.             }
  25.         }, 1, 10, TimeUnit.MINUTES);
  26.         if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) { 
  27.             // Register a listener to reload SslContext
  28.             try { 
  29.                 fileWatchService = new FileWatchService(
  30.                     new String[] { 
  31.                         TlsSystemConfig.tlsServerCertPath,
  32.                         TlsSystemConfig.tlsServerKeyPath,
  33.                         TlsSystemConfig.tlsServerTrustCertPath
  34.                     },
  35.                     new FileWatchService.Listener() { 
  36.                         boolean certChanged, keyChanged = false;
  37.                         @Override
  38.                         public void onChanged(String path) { 
  39.                             if (path.equals(TlsSystemConfig.tlsServerTrustCertPath)) { 
  40.                                 log.info("The trust certificate changed, reload the ssl context");
  41.                                 reloadServerSslContext();
  42.                             }
  43.                             if (path.equals(TlsSystemConfig.tlsServerCertPath)) { 
  44.                                 certChanged = true;
  45.                             }
  46.                             if (path.equals(TlsSystemConfig.tlsServerKeyPath)) { 
  47.                                 keyChanged = true;
  48.                             }
  49.                             if (certChanged && keyChanged) { 
  50.                                 log.info("The certificate and private key changed, reload the ssl context");
  51.                                 certChanged = keyChanged = false;
  52.                                 reloadServerSslContext();
  53.                             }
  54.                         }
  55.                         private void reloadServerSslContext() { 
  56.                             ((NettyRemotingServer) remotingServer).loadSslContext();
  57.                         }
  58.                     });
  59.             } catch (Exception e) { 
  60.                 log.warn("FileWatchService created error, can't load the certificate dynamically");
  61.             }
  62.         }
  64.         return true;
  65.     }

定时任务线程池

上面代码容易看懂的就是也是开了很多定时器任务线程，比如去打印Namespace的 k v配置
在这里插入图片描述
开启一个定时任务去扫描不活跃的broker，并剔除下线

  1. NamesrvController.this.routeInfoManager.scanNotActiveBroker();
  3. public void scanNotActiveBroker() { 
  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
  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.     }

构造Netty相关线程池

然后我们重点看看与Netty相关的代码

  1. this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
  3. public NettyRemotingServer(final NettyServerConfig nettyServerConfig,
  4.         final ChannelEventListener channelEventListener) { 
  5.         super(nettyServerConfig.getServerOnewaySemaphoreValue(), nettyServerConfig.getServerAsyncSemaphoreValue());
  6.         //用于启动NIO服务端的辅助启动类
  7.         this.serverBootstrap = new ServerBootstrap();
  8.         this.nettyServerConfig = nettyServerConfig;
  9.         this.channelEventListener = channelEventListener;
  11.         int publicThreadNums = nettyServerConfig.getServerCallbackExecutorThreads();
  12.         if (publicThreadNums <= 0) { 
  13.             publicThreadNums = 4;
  14.         }
  16.         this.publicExecutor = Executors.newFixedThreadPool(publicThreadNums, new ThreadFactory() { 
  17.             private AtomicInteger threadIndex = new AtomicInteger(0);
  19.             @Override
  20.             public Thread newThread(Runnable r) { 
  21.                 return new Thread(r, "NettyServerPublicExecutor_" + this.threadIndex.incrementAndGet());
  22.             }
  23.         });
  25.         if (useEpoll()) { 
  26.             this.eventLoopGroupBoss = new EpollEventLoopGroup(1, new ThreadFactory() { 
  27.                 private AtomicInteger threadIndex = new AtomicInteger(0);
  29.                 @Override
  30.                 public Thread newThread(Runnable r) { 
  31.                     return new Thread(r, String.format("NettyEPOLLBoss_%d", this.threadIndex.incrementAndGet()));
  32.                 }
  33.             });
  35.             this.eventLoopGroupSelector = new EpollEventLoopGroup(nettyServerConfig.getServerSelectorThreads(), new ThreadFactory() { 
  36.                 private AtomicInteger threadIndex = new AtomicInteger(0);
  37.                 private int threadTotal = nettyServerConfig.getServerSelectorThreads();
  39.                 @Override
  40.                 public Thread newThread(Runnable r) { 
  41.                     return new Thread(r, String.format("NettyServerEPOLLSelector_%d_%d", threadTotal, this.threadIndex.incrementAndGet()));
  42.                 }
  43.             });
  44.         } else { 
  45.             // 工作线程
  46.             this.eventLoopGroupBoss = new NioEventLoopGroup(1, new ThreadFactory() { 
  47.                 private AtomicInteger threadIndex = new AtomicInteger(0);
  49.                 @Override
  50.                 public Thread newThread(Runnable r) { 
  51.                     return new Thread(r, String.format("NettyNIOBoss_%d", this.threadIndex.incrementAndGet()));
  52.                 }
  53.             });
  55.             this.eventLoopGroupSelector = new NioEventLoopGroup(nettyServerConfig.getServerSelectorThreads(), new ThreadFactory() { 
  56.                 private AtomicInteger threadIndex = new AtomicInteger(0);
  57.                 private int threadTotal = nettyServerConfig.getServerSelectorThreads();
  59.                 @Override
  60.                 public Thread newThread(Runnable r) { 
  61.                     return new Thread(r, String.format("NettyServerNIOSelector_%d_%d", threadTotal, this.threadIndex.incrementAndGet()));
  62.                 }
  63.             });
  64.         }
  66.         loadSslContext();
  67.     }

可以看到这里主要用于初始化Netty的一些相关线程，其中也会判断NIO是使用通用的(Common)NIO实现(NioEventLoopGroup)，还是使用Linux封装的NIO实现(EpollEventLoopGroup)

可以看到这里只是创建了一些Netty相关的线程池，并没有启动Netty服务器。

启动Netty服务器

做了这么多准备工作，最后启动Netty服务器的实现还是这里启动的
在这里插入图片描述

  1. public void start() throws Exception { 
  2.         this.remotingServer.start();
  4.         if (this.fileWatchService != null) { 
  5.             this.fileWatchService.start();
  6.         }
  7.     }
  10. @Override
  11.     public void start() { 
  12.         this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
  13.             nettyServerConfig.getServerWorkerThreads(),
  14.             new ThreadFactory() { 
  16.                 private AtomicInteger threadIndex = new AtomicInteger(0);
  18.                 @Override
  19.                 public Thread newThread(Runnable r) { 
  20.                     return new Thread(r, "NettyServerCodecThread_" + this.threadIndex.incrementAndGet());
  21.                 }
  22.             });
  24.         prepareSharableHandlers();
  26.         ServerBootstrap childHandler =
  27.             this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
  28.                 .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
  29.                 .option(ChannelOption.SO_BACKLOG, 1024)//用于临时存放已完成三次握手的请求的队列的最大长度
  30.                 .option(ChannelOption.SO_REUSEADDR, true)
  31.                 .option(ChannelOption.SO_KEEPALIVE, false)// 是否启用心跳保活机制
  32.                 .childOption(ChannelOption.TCP_NODELAY, true)//启用或关于Nagle算法。如果要求高实时性，有数据发送时就马上发送，就将该选项设置为true关闭Nagle算法；如果要减少发送次数减少网络交互，就设置为false等累积一定大小后再发送。默认为false
  33.                 .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())// 定义传输的系统缓冲区buf的大小
  34.                 .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())// 定义接收的系统缓冲区buf的大小
  35.                 .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))
  36.                 .childHandler(new ChannelInitializer<SocketChannel>() { 
  37.                     @Override
  38.                     public void initChannel(SocketChannel ch) throws Exception { 
  39.                         ch.pipeline()
  40.                             .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME, handshakeHandler)
  41.                             .addLast(defaultEventExecutorGroup,
  42.                                 encoder,
  43.                                 new NettyDecoder(),
  44.                                 new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
  45.                                 connectionManageHandler,
  46.                                 serverHandler
  47.                             );
  48.                     }
  49.                 });
  51.         if (nettyServerConfig.isServerPooledByteBufAllocatorEnable()) { 
  52.             childHandler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
  53.         }
  55.         try { 
  56.             ChannelFuture sync = this.serverBootstrap.bind().sync();
  57.             InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
  58.             this.port = addr.getPort();
  59.         } catch (InterruptedException e1) { 
  60.             throw new RuntimeException("this.serverBootstrap.bind().sync() InterruptedException", e1);
  61.         }
  63.         if (this.channelEventListener != null) { 
  64.             this.nettyEventExecutor.start();
  65.         }
  67.         // 定时扫描请求超过1秒的请求
  68.         this.timer.scheduleAtFixedRate(new TimerTask() { 
  70.             @Override
  71.             public void run() { 
  72.                 try { 
  73.                     NettyRemotingServer.this.scanResponseTable();
  74.                 } catch (Throwable e) { 
  75.                     log.error("scanResponseTable exception", e);
  76.                 }
  77.             }
  78.         }, 1000 * 3, 1000);
  79.     }

这里都是Netty相关的代码，无非就是绑定一些编解码Handler，一些心跳、认证Handler等，再就是Socket相关的一些配置。

总结

至此NameServer计算启动完成了，然后监听端口9876，等待Broker注册，上面的一些Netty相关的代码代码我们并没有去重点分析，因为这算是Netty相关的部分。总之我们对NameServer的启动有了大致的了解，后续我们就看看Broker是如何注册到NameServer上去的，Broker是如何与NameServer进行通信的

发表评论

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

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

相关阅读