RocketMq-源码学习-二-Nameserv

nameserv是服务发现中心，底层采用netty监听

NamesrvStartup

服务的启动类，通过命令行参数初始化NamesrvController，并且启动NamesrvController。

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;
    }

    //初始化nameservconfig和netteyServerConfig
    final NamesrvConfig namesrvConfig = new NamesrvConfig();
    final NettyServerConfig nettyServerConfig = new NettyServerConfig();

    ...

    //关键代码，创建NamesrvController
    final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);

    // remember all configs to prevent discard
    controller.getConfiguration().registerConfig(properties);
    return controller;
}

start方法

public static NamesrvController start(final NamesrvController controller) throws Exception {
    ......

    //初始化配置
    boolean initResult = controller.initialize();

    ......

    Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {
        @Override
        public Void call() throws Exception {
            controller.shutdown();
            return null;
        }
    }));

    //关键代码start
    controller.start();

    return controller;
}

NamesrvController

构造函数

public NamesrvController(NamesrvConfig namesrvConfig, NettyServerConfig nettyServerConfig) {
    //nameserver的配置
    this.namesrvConfig = namesrvConfig;
    //netty相关的配置
    this.nettyServerConfig = nettyServerConfig;
    //kv存储的管理类
    this.kvConfigManager = new KVConfigManager(this);
    //broker topic的信息管理
    this.routeInfoManager = new RouteInfoManager();
    //broker心跳的类，是ChannelEventListener的实现类，Netty的事件触发后处理事件的详细的类，具体调用见NettyConnectManageHandler
    this.brokerHousekeepingService = new BrokerHousekeepingService(this);
    this.configuration = new Configuration(
        log,
        this.namesrvConfig, this.nettyServerConfig
    );
    //storePath的保存
    this.configuration.setStorePathFromConfig(this.namesrvConfig, "configStorePath");
}

initialize

public boolean initialize() {

       //kv设置load到内存
       this.kvConfigManager.load();

       //核心：创建nettyserver
       this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);

       //远程命令处理类,具体调用在NettyRemotingAbstract的processMessageReceived里
       this.remotingExecutor =
           Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
       this.registerProcessor();

       //扫描未活跃的broker并且移除他们每10秒一次
       this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

           @Override
           public void run() {
               NamesrvController.this.routeInfoManager.scanNotActiveBroker();
           }
       }, 5, 10, TimeUnit.SECONDS);

       //打印所有kvstore的kv
       this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

           @Override
           public void run() {
               NamesrvController.this.kvConfigManager.printAllPeriodically();
           }
       }, 1, 10, TimeUnit.MINUTES);

       //tls--监控tls的文件状态如果修改了重新reload--一个私有方法reloadServerSslContext
       if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {
           // Register a listener to reload SslContext
           try {
               fileWatchService = new FileWatchService(
                   ......见源码
               );
           } catch (Exception e) {
               log.warn("FileWatchService created error, can't load the certificate dynamically");
           }
       }

       return true;
   }

start

public void start() throws Exception {
    //start servbice
    this.remotingServer.start();

    //监听ssl的变化，如果有变化remotingServer.loadSslContext()
    if (this.fileWatchService != null) {
        this.fileWatchService.start();
    }
}

remotingServer–NettyRemotingServer

@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());
            }
        });

    ServerBootstrap childHandler =
        this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)//bossgruop和workgroup
            .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
            .option(ChannelOption.SO_BACKLOG, 1024)//listerner的长度，过小会导致tcp链接简历失败
            .option(ChannelOption.SO_REUSEADDR, true)//服务器端口可被重用
            .option(ChannelOption.SO_KEEPALIVE, false)
            .childOption(ChannelOption.TCP_NODELAY, true)//禁用Nagle算法，有效提高网络负载
            .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())//发送缓冲区大小
            .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())//接受缓冲区大小
            .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))//设置端口
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    //注意：使用第三个EventGroup,defaultEventExecutorGroup，建立链接、读写任务分开
                    ch.pipeline()
                        .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME,
                            new HandshakeHandler(TlsSystemConfig.tlsMode))
                        .addLast(defaultEventExecutorGroup,
                            new NettyEncoder(),//MessageToByteEncoder
                            new NettyDecoder(),//LengthFieldBasedFrameDecoder具体见netty,解决粘包问题
                            new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),//心跳检测机制
                            new NettyConnectManageHandler(),//处理io链接相关的事件注意啊channelActive,channelInActive,userEventTriggered（心跳）,exceptionCaught
                            new NettyServerHandler()//处理request和response信息
                        );
                }
            });

    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);
    }

    //netty的事件处理类
    // 1.nettyEventExecutor是一个线程，run里是死循环，nettyEventExecutor维护了个阻塞队列，采用生产者消费者模式。putNettEvent放事件，线程异步消费事件
    // 2.事件来了后，通过NettyRemotingServer.this.channelEventListener方法类处理事件
    if (this.channelEventListener != null) {
        this.nettyEventExecutor.start();
    }

    this.timer.scheduleAtFixedRate(new TimerTask() {
        @Override
        public void run() {
            try {
                NettyRemotingServer.this.scanResponseTable();
            } catch (Throwable e) {
                log.error("scanResponseTable exception", e);
            }
        }
    }, 1000 * 3, 1000);
}