RocketMq总决式-NameServer源码1（通过源码看配置）

落日映苍穹つ 2022-11-18 13:35 133阅读 0赞

# 命名服务器启动 #

## 一张图说NameServer启动流程 ##

本节我们一张图通过源码看配置  
下一节我们一张图通过源码看路由管理 ：RocketMq总决式-NameServer源码2（通过源码看路由管理）  
![NameServer启动说明][NameServer]

> 上图表示了NameServer的启动调用逻辑，其中包含了两个重要的配置对象`NamesrvConfig`，`NettyServerConfig` ，和NameServer的核心控制器`NamesrvController`,NameServer的核心功能在NameController做实现，例如：路由管理。

## 一起读源码吧 ##

在筑基篇介绍了RocketMQ的集群部署方法，具体可以参考 [RocketMQ筑基篇-部署集群（Centos7）][RocketMQ_-_Centos7]。  
其中NameServer的启动命令为：

nohup sh bin/mqnamesrv > nameServer-start.log &

那么我们看一下这个`mqnamesrv` 脚本文件内容

#!/bin/sh
    if [ -z "$ROCKETMQ_HOME" ] ; then
      ## resolve links - $0 may be a link to maven's home
      PRG="$0"
      # need this for relative symlinks
      while [ -h "$PRG" ] ; do
        ls=`ls -ld "$PRG"`
        link=`expr "$ls" : '.*-> $.*$$'`
        if expr "$link" : '/.*' > /dev/null; then
          PRG="$link"
        else
          PRG="`dirname "$PRG"`/$link"
        fi
      done
      saveddir=`pwd`
      ROCKETMQ_HOME=`dirname "$PRG"`/..
      # make it fully qualified
      ROCKETMQ_HOME=`cd "$ROCKETMQ_HOME" && pwd`
      cd "$saveddir"
    fi
    export ROCKETMQ_HOME
    sh ${ROCKETMQ_HOME}/bin/runserver.sh org.apache.rocketmq.namesrv.NamesrvStartup $@

直接来到最后一行，可以看到 NameServer的启动类正是 `org.apache.rocketmq.namesrv.NamesrvStartup`

## NameServer启动类详解 ##

> NameServer使用了Netty实现，这也是RocketMQ高性能的一个原因。  
> NameServer命令行使用了Apache的commons工具集中的`cli`实现（如果对cli有兴趣可以在评论区留言）。

这里笔者为源代码添加了一些注释，以便自己和读者更方便的阅读RocketMQ源代码，博文仅展示主要代码注释，更多注释请参考[笔者github][github] https://github.com/keepgoon/rocketmq/

public class NamesrvStartup { 
        public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException { 
            // 设置RocketMQ的版本（因此如果想知道启动的NameServer版本可以通过系统属性进行查询）
            System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
    
            Options options = ServerUtil.buildCommandlineOptions(new Options());
            commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());
            if (null == commandLine) { 
                System.exit(-1);
                return null;
            }
            // NameServer的核心配置对象
            final NamesrvConfig namesrvConfig = new NamesrvConfig();
            // NameServer的Netty配置对象（包括端口，线程数量等）
            final NettyServerConfig nettyServerConfig = new NettyServerConfig();
            // 默认端口
            nettyServerConfig.setListenPort(9876);
            final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
            // remember all configs to prevent discard
            controller.getConfiguration().registerConfig(properties);
            return controller;
        }
    	// 调用名称控制器，启动名称控制器
        public static NamesrvController start(final NamesrvController controller) throws Exception {  
        	// 调用名称控制器的初始化
            boolean initResult = controller.initialize();
            if (!initResult) { 
                controller.shutdown();
                System.exit(-3);
            }
    
            Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() { 
                @Override
                public Void call() throws Exception { 
                    controller.shutdown();
                    return null;
                }
            }));
    
            controller.start();
            return controller;
        }
    }

1.  首先启动时先设置了系统版本号，并且加载启动命令以及参数。其中命令处理工具使用了 apache的 `cli` 。深入读可看出，h 为help ,会调用配置对象打印所有配置，c为配置文件，n为NameServerAddr命名服务地址列表：可以通过 -c 指定配置文件位置，也可以通过 --listenPort==8888修改某一项配置。
2.  NettyServerConfig 声明了Netty服务的配置信息，可以看出默认端口是9876。具体配置项见 `NettyServerConfig`
3.  然后调用了名称控制器（NamesrvController ）的初始化方法，这个方法是启动的核心，NamesrvController 也是NameServer的核心对象，路由管理就是在这里进行的，我们下一章见 通过源码看路由管理
4.  最后调用start方法，启动名称服务

## NettyServerConfig配置详情 ##

通过源码读配置的核心来了，这就是核心配置对象了，具体详见：

public class NettyServerConfig implements Cloneable { 
        /** Nettp 服务端口 */
        private int listenPort = 8888;
        /** 工具工作线程数量 */
        private int serverWorkerThreads = 8;
        /** Netty 回调任务线程数量。处理消息发送，消息消费，心跳检查等 */
        private int serverCallbackExecutorThreads = 0;
        /** IO 线程池线程个数，主要是 NameServer、Broker 端解析请求、返回响应的线程个数，这类线程池主要是处理网络请求的，解析请求包，然后转发到各个业务线程池完成具体的业务操作，然后将结果再返回调用方 */
        private int serverSelectorThreads = 3;
        /** 单向请求并发度 （Broker端） */
        private int serverOnewaySemaphoreValue = 256;
        /** 异步请求并发度 （Broker端） */
        private int serverAsyncSemaphoreValue = 64;
        /** 网络最大空闲时间，单位秒。如果链接超过这个时间，链接会断开 */
        private int serverChannelMaxIdleTimeSeconds = 120;
        /** 网络套接字发送缓存的大小，默认64K */
        private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;
        /** 网络套接字接收缓存大小，默认64K */
        private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;
        /** 是否开启缓存，这个建议开启 */
        private boolean serverPooledByteBufAllocatorEnable = true;
    
        /** Epoll 无聊的小知识又要增加了 ，首先linux系统才有这个东西，它是Linux处理高并发的，类似netty非阻塞，多路复用，linux下建议开启 */
        private boolean useEpollNativeSelector = false;
    
    }

## NamesrvController 的初始化方法 ##

package org.apache.rocketmq.namesrv;
    
    public class NamesrvController { 
        /** * 初始化名称控制器 * * @return boolean **/
        public boolean initialize() { 
            // 加载key-value配置管理
            this.kvConfigManager.load();
            // 创建一个远程处理服务，该服务实现在remoting包中，这里仅仅是船机一个对象，并没有启动服务，服务启动在start中进行
            this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
            // 线程池实际是 JDK并发包中的ExecutorService
            this.remotingExecutor =
                Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
            // 注册处理器
            this.registerProcessor();
            // 开启一个延迟5秒，没十秒执行一次的定时任务 扫码宕机的Broker
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    
                @Override
                public void run() { 
                    NamesrvController.this.routeInfoManager.scanNotActiveBroker();
                }
            }, 5, 10, TimeUnit.SECONDS);
            // 开启一个延迟1分钟，没10分钟执行一次的定时任务，打印所有的Key-Value配置
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    
                @Override
                public void run() { 
                    NamesrvController.this.kvConfigManager.printAllPeriodically();
                }
            }, 1, 10, TimeUnit.MINUTES);
            // 如果开启了TLS，需要处理安全协议
            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;
        }
    
        private void registerProcessor() { 
            if (namesrvConfig.isClusterTest()) { 
    
                this.remotingServer.registerDefaultProcessor(new ClusterTestRequestProcessor(this, namesrvConfig.getProductEnvName()),
                    this.remotingExecutor);
            } else { 
    
                this.remotingServer.registerDefaultProcessor(new DefaultRequestProcessor(this), this.remotingExecutor);
            }
        }
    
    
    }

[NameServer]: /images/20221022/52a0125dbbc54610a1c5bf1207e4b7c5.png
[RocketMQ_-_Centos7]: https://keepgoon.blog.csdn.net/article/details/111825767
[github]: https://github.com/keepgoon/rocketmq/