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

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

### 文章目录 ###

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

## 入口 ##

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

NamesrvController controller = createNamesrvController(args);
    	start(controller);

## NamesrvController是如何被创建出来的 ##

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

public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException { 
            System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
            //PackageConflictDetect.detectFastjson();
    
            Options options = ServerUtil.buildCommandlineOptions(new Options());
            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();
            nettyServerConfig.setListenPort(9876);
            // 启动namesrv 参数带c代表读取配置文件
            if (commandLine.hasOption('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);
    
                    namesrvConfig.setConfigStorePath(file);
    
                    System.out.printf("load config properties file OK, %s%n", file);
                    in.close();
                }
            }
    
            // 如果启动参数带p 打印NameSrv启动参数
            if (commandLine.hasOption('p')) { 
                InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
                MixAll.printObjectProperties(console, namesrvConfig);
                MixAll.printObjectProperties(console, nettyServerConfig);
                System.exit(0);
            }
            // 获取配置类容填充到namesrvConfig
            MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);
            // 如果rocketmqHome 打印异常退出
            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();
            configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");
            // 打印NameServer所有配置信息
            log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    
            MixAll.printObjectProperties(log, namesrvConfig);
            MixAll.printObjectProperties(log, nettyServerConfig);
    
            final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
    
            // remember all configs to prevent discard
            controller.getConfiguration().registerConfig(properties);
    
            return controller;
        }

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

## 启动NameServer ##

启动NameServer的核心入口方法是

start(controller);

我们看看这段代码的实现

public static NamesrvController start(final NamesrvController controller) throws Exception { 
    
            if (null == controller) { 
                throw new IllegalArgumentException("NamesrvController is null");
            }
            // 初始化核心逻辑
            boolean initResult = controller.initialize();
            if (!initResult) { 
                controller.shutdown();
                System.exit(-3);
            }
    
            // 注册钩子函数 在JVM进程关闭时，优雅地释放netty服务、线程池等资源
            Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() { 
                @Override
                public Void call() throws Exception { 
                    controller.shutdown();
                    return null;
                }
            }));
            controller.start();
            return controller;
        }

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

boolean initResult = controller.initialize();

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

public boolean initialize() { 
    
            this.kvConfigManager.load();
            // 构建Netty服务器
            this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
            // netty 工作线程
            this.remotingExecutor =           Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
            
            // 将工作线程池给netty
            this.registerProcessor();
            // 定时检测不活跃的Broker
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
                @Override
                public void run() { 
                    NamesrvController.this.routeInfoManager.scanNotActiveBroker();
                }
            }, 5, 10, TimeUnit.SECONDS);
    
            // 创建定时任务--每个10分钟打印一遍KV配置
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
                @Override
                public void run() { 
                    NamesrvController.this.kvConfigManager.printAllPeriodically();
                }
            }, 1, 10, TimeUnit.MINUTES);
            if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) { 
                // Register a listener to reload SslContext
                try { 
                    fileWatchService = new FileWatchService(
                        new String[] { 
                            TlsSystemConfig.tlsServerCertPath,
                            TlsSystemConfig.tlsServerKeyPath,
                            TlsSystemConfig.tlsServerTrustCertPath
                        },
                        new FileWatchService.Listener() { 
                            boolean certChanged, keyChanged = false;
                            @Override
                            public void onChanged(String path) { 
                                if (path.equals(TlsSystemConfig.tlsServerTrustCertPath)) { 
                                    log.info("The trust certificate changed, reload the ssl context");
                                    reloadServerSslContext();
                                }
                                if (path.equals(TlsSystemConfig.tlsServerCertPath)) { 
                                    certChanged = true;
                                }
                                if (path.equals(TlsSystemConfig.tlsServerKeyPath)) { 
                                    keyChanged = true;
                                }
                                if (certChanged && keyChanged) { 
                                    log.info("The certificate and private key changed, reload the ssl context");
                                    certChanged = keyChanged = false;
                                    reloadServerSslContext();
                                }
                            }
                            private void reloadServerSslContext() { 
                                ((NettyRemotingServer) remotingServer).loadSslContext();
                            }
                        });
                } catch (Exception e) { 
                    log.warn("FileWatchService created error, can't load the certificate dynamically");
                }
            }
    
            return true;
        }

### 定时任务线程池 ###

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

NamesrvController.this.routeInfoManager.scanNotActiveBroker();
    
    public void scanNotActiveBroker() { 
            Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
            while (it.hasNext()) { 
                Entry<String, BrokerLiveInfo> next = it.next();
                long last = next.getValue().getLastUpdateTimestamp();
                // 当前时间超过上一次心跳时间 + 默认120s
                if ((last + BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) { 
                    RemotingUtil.closeChannel(next.getValue().getChannel());
                    it.remove();
                    log.warn("The broker channel expired, {} {}ms", next.getKey(), BROKER_CHANNEL_EXPIRED_TIME);
                    // 任务Broker下线 移除掉Broker
                    this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
                }
            }
        }

### 构造Netty相关线程池 ###

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

this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
    
    public NettyRemotingServer(final NettyServerConfig nettyServerConfig,
            final ChannelEventListener channelEventListener) { 
            super(nettyServerConfig.getServerOnewaySemaphoreValue(), nettyServerConfig.getServerAsyncSemaphoreValue());
            //用于启动NIO服务端的辅助启动类
            this.serverBootstrap = new ServerBootstrap();
            this.nettyServerConfig = nettyServerConfig;
            this.channelEventListener = channelEventListener;
    
            int publicThreadNums = nettyServerConfig.getServerCallbackExecutorThreads();
            if (publicThreadNums <= 0) { 
                publicThreadNums = 4;
            }
    
            this.publicExecutor = Executors.newFixedThreadPool(publicThreadNums, new ThreadFactory() { 
                private AtomicInteger threadIndex = new AtomicInteger(0);
    
                @Override
                public Thread newThread(Runnable r) { 
                    return new Thread(r, "NettyServerPublicExecutor_" + this.threadIndex.incrementAndGet());
                }
            });
    
            if (useEpoll()) { 
                this.eventLoopGroupBoss = new EpollEventLoopGroup(1, new ThreadFactory() { 
                    private AtomicInteger threadIndex = new AtomicInteger(0);
    
                    @Override
                    public Thread newThread(Runnable r) { 
                        return new Thread(r, String.format("NettyEPOLLBoss_%d", this.threadIndex.incrementAndGet()));
                    }
                });
    
                this.eventLoopGroupSelector = new EpollEventLoopGroup(nettyServerConfig.getServerSelectorThreads(), new ThreadFactory() { 
                    private AtomicInteger threadIndex = new AtomicInteger(0);
                    private int threadTotal = nettyServerConfig.getServerSelectorThreads();
    
                    @Override
                    public Thread newThread(Runnable r) { 
                        return new Thread(r, String.format("NettyServerEPOLLSelector_%d_%d", threadTotal, this.threadIndex.incrementAndGet()));
                    }
                });
            } else { 
                // 工作线程
                this.eventLoopGroupBoss = new NioEventLoopGroup(1, new ThreadFactory() { 
                    private AtomicInteger threadIndex = new AtomicInteger(0);
    
                    @Override
                    public Thread newThread(Runnable r) { 
                        return new Thread(r, String.format("NettyNIOBoss_%d", this.threadIndex.incrementAndGet()));
                    }
                });
    
                this.eventLoopGroupSelector = new NioEventLoopGroup(nettyServerConfig.getServerSelectorThreads(), new ThreadFactory() { 
                    private AtomicInteger threadIndex = new AtomicInteger(0);
                    private int threadTotal = nettyServerConfig.getServerSelectorThreads();
    
                    @Override
                    public Thread newThread(Runnable r) { 
                        return new Thread(r, String.format("NettyServerNIOSelector_%d_%d", threadTotal, this.threadIndex.incrementAndGet()));
                    }
                });
            }
    
            loadSslContext();
        }

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

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

### 启动Netty服务器 ###

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

public void start() throws Exception { 
            this.remotingServer.start();
    
            if (this.fileWatchService != null) { 
                this.fileWatchService.start();
            }
        }
    
    
    @Override
        public void start() { 
            this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
                nettyServerConfig.getServerWorkerThreads(),
                new ThreadFactory() { 
    
                    private AtomicInteger threadIndex = new AtomicInteger(0);
    
                    @Override
                    public Thread newThread(Runnable r) { 
                        return new Thread(r, "NettyServerCodecThread_" + this.threadIndex.incrementAndGet());
                    }
                });
    
            prepareSharableHandlers();
    
            ServerBootstrap childHandler =
                this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
                    .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)//用于临时存放已完成三次握手的请求的队列的最大长度
                    .option(ChannelOption.SO_REUSEADDR, true)
                    .option(ChannelOption.SO_KEEPALIVE, false)// 是否启用心跳保活机制
                    .childOption(ChannelOption.TCP_NODELAY, true)//启用或关于Nagle算法。如果要求高实时性，有数据发送时就马上发送，就将该选项设置为true关闭Nagle算法；如果要减少发送次数减少网络交互，就设置为false等累积一定大小后再发送。默认为false
                    .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())// 定义传输的系统缓冲区buf的大小
                    .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())// 定义接收的系统缓冲区buf的大小
                    .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))
                    .childHandler(new ChannelInitializer<SocketChannel>() { 
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception { 
                            ch.pipeline()
                                .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME, handshakeHandler)
                                .addLast(defaultEventExecutorGroup,
                                    encoder,
                                    new NettyDecoder(),
                                    new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                                    connectionManageHandler,
                                    serverHandler
                                );
                        }
                    });
    
            if (nettyServerConfig.isServerPooledByteBufAllocatorEnable()) { 
                childHandler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
            }
    
            try { 
                ChannelFuture sync = this.serverBootstrap.bind().sync();
                InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
                this.port = addr.getPort();
            } catch (InterruptedException e1) { 
                throw new RuntimeException("this.serverBootstrap.bind().sync() InterruptedException", e1);
            }
    
            if (this.channelEventListener != null) { 
                this.nettyEventExecutor.start();
            }
    
            // 定时扫描请求超过1秒的请求
            this.timer.scheduleAtFixedRate(new TimerTask() { 
    
                @Override
                public void run() { 
                    try { 
                        NettyRemotingServer.this.scanResponseTable();
                    } catch (Throwable e) { 
                        log.error("scanResponseTable exception", e);
                    }
                }
            }, 1000 * 3, 1000);
        }

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

## 总结 ##

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

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2VpaHViZWF0cw_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/06367852c8304dcdaf0e11423896e452.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2VpaHViZWF0cw_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220828/b12405b6a10b4c658f6caded4714ee6d.png
[637c8fb97cf74bed9377316f464dca37.png]: /images/20220828/ab26b8e8d8574629ac913cb2c80524a3.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2VpaHViZWF0cw_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220828/5ff205b8a1d143f49b053f5501bd8b00.png