RocketMQ源码--4--Namesrv启动过程

桃扇骨 2022-10-18 11:29 45阅读 0赞

# `RocketMQ`源码–`4`–`Namesrv`启动过程 #

### 文章目录 ###

*  \`RocketMQ\`源码--\`4\`--\`Namesrv\`启动过程
 *  1 源码启动\`Namesrv\`
 *  2 \`Namesrv\`启动过程
 *   *  2.1 创建\`NamesrvController\`
     *   *  2.1.1 初始化配置对象\`NamesrvConfig\`
         *  2.1.2 初始化配置对象\`NettyServerConfig\`
         *  2.1.3 \`new\`一个\`NamesrvController\`对象
     *  2.2 启动\`Namesrv\`
     *   *  2.2.1 初始化\`NamesrvController\`
         *  2.2.2 启动\`NamesrvController\`
 *  3 \`KVConfigManager\`组件
 *   *  3.1 构造方法
     *  3.2 \`load()\`方法，解析\`kvConfig.json\`文件
     *   *  3.2.1 获取\`kvConfig.json\`文件内容
         *  3.2.2 \`KVConfigSerializeWrapper\`
     *  3.3 \`printAllPeriodically()\`方法，打印\`configTable\`的内容
 *  4 \`RouteInfoManager\`组件
 *   *  4.1 构造方法
     *  4.2 \`scanNotActiveBroker()\`方法，扫描并剔除不活跃的\`Broker\`路由信息
     *  4.3 \`onChannelDestroy()\`方法，剔除其他路由表中宕机节点信息
 *  5 \`BrokerHousekeepingService\`组件

# 1 源码启动`Namesrv` #

>  源码启动其实挺简单的，首先你去`RocketMQ`官网，将源码包下载下来，然后使用`idea`打开，直接启动，你会发现控制台提示你需要配置一个环境变量`ROCKETMQ_HOME`，所以这时候，我们就直接在`idea`中配置一个环境变量`ROCKETMQ_HOME`，并将它的值指定为当前项目的根路径，比如我的配置就是`ROCKETMQ_HOME=D:\java学习\rocketmq\rocketmq-release-4.7.0`

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMDEyOTgx_size_16_color_FFFFFF_t_70_pic_center]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMDEyOTgx_size_16_color_FFFFFF_t_70_pic_center 1]

这样配置好了之后，接下来就需要将配置文件复制过来了。

直接在`ROCKETMQ_HOME`指定的路径下面新建一个文件夹`conf`，然后将二进制包中`conf`目录下的`logback_namesrv.xml`配置文件拷贝到你新建的文件夹`conf`目录下。

最后直接启动就`ok`了。

# 2 `Namesrv`启动过程 #

>  `Namesrv`启动类的完全限定名为`org.apache.rocketmq.namesrv.NamesrvStartup`,我们从这个类的`main()`方法开始跟踪，看看启动过程中做了哪些事情？

public static void main(String[] args) { 
        // 调用下面的main0()方法启动
        main0(args);
    }
    
    public static NamesrvController main0(String[] args) { 
    
        try { 
            // 创建NamesrvController，见2.1
            NamesrvController controller = createNamesrvController(args);
            // 启动NamesrvController，见2.2
            start(controller);
            String tip = "The Name Server boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
            log.info(tip);
            System.out.printf("%s%n", tip);
            return controller;
        } catch (Throwable e) { 
            e.printStackTrace();
            System.exit(-1);
        }
    
        return null;
    }

> 总体来说就两个步骤
> 
> 1.  创建`NamesrvController`
> 2.  启动`NamesrvController`

## 2.1 创建`NamesrvController` ##

public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException { 
        // REMOTING_VERSION_KEY = "rocketmq.remoting.version" 保存当前RocketMQ的版本号
        System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
        //PackageConflictDetect.detectFastjson();
    
        // 命令行参数mqnamesrv解析
        Options options = ServerUtil.buildCommandlineOptions(new Options());
        commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());
        if (null == commandLine) { 
            System.exit(-1);
            return null;
        }
    
        // 创建一个Namesrv配置文件类对象
        final NamesrvConfig namesrvConfig = new NamesrvConfig();
        // 创建一个netty服务端配置文件类对象
        final NettyServerConfig nettyServerConfig = new NettyServerConfig();
        // 设置通信端口号为9876
        nettyServerConfig.setListenPort(9876);
        
        // 命令行参数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解析
        if (commandLine.hasOption('p')) { 
            InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
            MixAll.printObjectProperties(console, namesrvConfig);
            MixAll.printObjectProperties(console, nettyServerConfig);
            System.exit(0);
        }
    
        // 还是命令行参数解析，这个先不管，我们先看主流程
        MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);
    
        // 判断是否配置了ROCKETMQ_HOME这个环境变量，没有配置，在此处就会报错
        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);
        }
    
        // 日志配置，可以发现日志的配置文件是环境变量ROCKETMQ_HOME下面conf目录
        // 下面的logback_namesrv.xml文件，这是固定的
        LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
        JoranConfigurator configurator = new JoranConfigurator();
        configurator.setContext(lc);
        lc.reset();
        configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");
    
        log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    
        // 打印日志相关配置信息
        MixAll.printObjectProperties(log, namesrvConfig);
        MixAll.printObjectProperties(log, nettyServerConfig);
    
        // 直接new一个NamesrvController对象
        final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
    
        // remember all configs to prevent discard
        // 记住所有配置，防止丢失
        controller.getConfiguration().registerConfig(properties);
    
        return controller;
    }

> 总结一下这个方法的流程：
> 
> 1.  解析命令行参数`mqnamesrv`
> 2.  初始化两个配置对象（`NamesrvConfig`、`NettyServerConfig`）
> 3.  解析命令行参数`c`
> 4.  解析命令行参数`p`
> 5.  解析`conf/logback_namesrv.xml`文件，应用指定配置生成`log`对象
> 6.  根据两个配置对象生成一个`NamesrvController`对象
> 7.  记住配置，防止丢失
> 
> 梳理一下这个流程，还是挺简单的，我们重点看一下`2`和`6`。

### 2.1.1 初始化配置对象`NamesrvConfig` ###

> `NamesrvConfig`只有一个默认的构造方法，但是类中的几个字段都指定了默认值

public class NamesrvConfig { 
        private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    
        // ROCKETMQ_HOME_PROPERTY = "rocketmq.home.dir"
        // ROCKETMQ_HOME_ENV = "ROCKETMQ_HOME"
        // System.getProperty(key,def)这个方法会先去系统中找key对应的值，如果为null，则
        // 返回后面def值，其实def就相当于一个默认值
        private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
        // separator = "\" 
        // 默认为C:\Users\Administrator\namesrv\kvConfig.json
        private String kvConfigPath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "kvConfig.json";
        // 默认为C:\Users\Administrator\namesrv\namesrv.properties
        private String configStorePath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "namesrv.properties";
        private String productEnvName = "center";
        private boolean clusterTest = false;
        private boolean orderMessageEnable = false;
    }

> 在创建对象的时候，所有的字段全部已经给定了初始值

### 2.1.2 初始化配置对象`NettyServerConfig` ###

> `NettyServerConfig`同样只有一个默认的构造方法，且类中的几个字段都指定了默认值

public class NettyServerConfig implements Cloneable { 
        // 后面会调用set方法，将其修改为9876
        private int listenPort = 8888;
        private int serverWorkerThreads = 8;
        private int serverCallbackExecutorThreads = 0;
        private int serverSelectorThreads = 3;
        private int serverOnewaySemaphoreValue = 256;
        private int serverAsyncSemaphoreValue = 64;
        private int serverChannelMaxIdleTimeSeconds = 120;
    
        private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;
        private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;
        private boolean serverPooledByteBufAllocatorEnable = true;
    
        /** * make make install * * * ../glibc-2.10.1/configure \ --prefix=/usr \ --with-headers=/usr/include \ * --host=x86_64-linux-gnu \ --build=x86_64-pc-linux-gnu \ --without-gd */
        private boolean useEpollNativeSelector = false;
    }

> 在创建对象的时候，所有的字段全部已经给定了初始值，这些配置都是`netty`的配置。

### 2.1.3 `new`一个`NamesrvController`对象 ###

> 我们来看一下这个构造方法

private final NamesrvConfig namesrvConfig;
    private final NettyServerConfig nettyServerConfig;
    
    // 初始化了一个只有单个线程的定时线程池
    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl(
        "NSScheduledThread"));
    
    // 加载kvConfig.json文件，并保存配置，见3
    private final KVConfigManager kvConfigManager;
    
    // Broker路由信息管理器，见4
    private final RouteInfoManager routeInfoManager;
    
    // 监听器，用来剔除连接关闭，连接异常的`Broker`节点的路由信息，见5
    private BrokerHousekeepingService brokerHousekeepingService;
    
    
    private Configuration configuration;
    
    
    public NamesrvController(NamesrvConfig namesrvConfig, NettyServerConfig nettyServerConfig) { 
        // 保存将两个配置对象
        this.namesrvConfig = namesrvConfig;
        this.nettyServerConfig = nettyServerConfig;
        // 创建一个KVConfigManager对象，见3.1
        this.kvConfigManager = new KVConfigManager(this);
        // 创建一个RouteInfoManager对象，见4.1
        this.routeInfoManager = new RouteInfoManager();
        this.brokerHousekeepingService = new BrokerHousekeepingService(this);
        this.configuration = new Configuration(
            log,
            this.namesrvConfig, this.nettyServerConfig
        );
        this.configuration.setStorePathFromConfig(this.namesrvConfig, "configStorePath");
    }

> 这上面涉及到很多的类，等后面用到了再看源码。

## 2.2 启动`Namesrv` ##

public static NamesrvController start(final NamesrvController controller) throws Exception { 
    
        if (null == controller) { 
            throw new IllegalArgumentException("NamesrvController is null");
        }
    
        // 初始化NamesrvController，见2.2.1
        boolean initResult = controller.initialize();
        // 未初始化成功就关闭程序
        if (!initResult) { 
            controller.shutdown();
            System.exit(-3);
        }
    
        // 添加一个钩子，关闭程序的时候回调shutdown()方法， 释放资源
        Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() { 
            @Override
            public Void call() throws Exception { 
                controller.shutdown();
                return null;
            }
        }));
    
        // 启动NamesrvController，见2.2.2
        controller.start();
    
        return controller;
    }

> 这个`start()`方法核心就干了两件事情
> 
> 1.  调用`initialize()`方法，初始化`NamesrvController`
> 2.  调用`start()`方法，启动`NamesrvController`

### 2.2.1 初始化`NamesrvController` ###

// NamesrvController对象创建的时候给定的初始值
    private final KVConfigManager kvConfigManager;
    
    // netty服务端对象
    private RemotingServer remotingServer;
    
    public boolean initialize() { 
        // 加载解析C:\Users\Administrator\namesrv\kvConfig.json文件，见3.2
        // 文件位置来源见2.1.1
        this.kvConfigManager.load();
    
        // 创建netty服务端对象
        this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
    
        // 工作线程池，线程数量由NettyServerConfig指定
        // 这个线程池主要用来处理Broker客户端的连接，Broker客户端需要定时上报它的地址信息（心跳）
        this.remotingExecutor =
            Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
    
        // netty相关，注册Processor
        this.registerProcessor();
    
        // 这个线程池会定时的去扫描超时的Broker，将其剔除路由表
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    
            @Override
            public void run() { 
                // 调用当前NamesrvController对象中routeInfoManager对象的
                // scanNotActiveBroker()方法，扫描活跃的Broker，见4.2
                NamesrvController.this.routeInfoManager.scanNotActiveBroker();
            }
        }, 5, 10, TimeUnit.SECONDS);
    
        // 又是一个定时任务，定时打印日志
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    
            @Override
            public void run() { 
                // 日志输出configTable的内容，见3.3
                NamesrvController.this.kvConfigManager.printAllPeriodically();
            }
        }, 1, 10, TimeUnit.MINUTES);
    
        // ssl相关，跳过
        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;
    }

> 总结该方法做的事情：
> 
> 1.  加载解析`C:\Users\Administrator\namesrv\kvConfig.json`文件
> 2.  创建`netty`服务端对象
> 3.  创建线程池，主要用来处理Broker客户端的连接
> 4.  设置两个定时任务
>     
>      *  定时的去扫描超时的`Broker`，将其剔除路由表
>      *  定时打印日志，`kvConfig.json`文件的内容

### 2.2.2 启动`NamesrvController` ###

// netty服务端对象
    private RemotingServer remotingServer;
    
    public void start() throws Exception { 
        // 启动netty服务器端
        this.remotingServer.start();
    
        // 启动FileWatchService
        if (this.fileWatchService != null) { 
            this.fileWatchService.start();
        }
    }

> 有关`netty`的知识，我会单独用一篇博客记录，我现在也不是很懂。

# 3 `KVConfigManager`组件 #

用途：加载`kvConfig.json`文件，并保存该文件中指定的配置。

## 3.1 构造方法 ##

private final NamesrvController namesrvController;
    
    public KVConfigManager(NamesrvController namesrvController) { 
        this.namesrvController = namesrvController;
    }

> 这个构造方法没什么内容，只是将`NamesrvController`对象的引用保存起来

## 3.2 `load()`方法，解析`kvConfig.json`文件 ##

// 配置表，专用来存储kvConfig.json文件中的配置
    private final HashMap<String/* Namespace */, HashMap<String/* Key */, String/* Value */>> configTable =
        new HashMap<String, HashMap<String, String>>();
    
    
    public void load() { 
        String content = null;
        try { 
            // 加载C:\Users\Administrator\namesrv\kvConfig.json得到文件内容，见3.2.1
            content = MixAll.file2String(this.namesrvController.getNamesrvConfig().getKvConfigPath());
        } catch (IOException e) { 
            log.warn("Load KV config table exception", e);
        }
        // 有内容，则进一步解析，格式化json字符串，json转KVConfigSerializeWrapper对象
        if (content != null) { 
            // 内部是fastJson来完成json字符串转换的
            KVConfigSerializeWrapper kvConfigSerializeWrapper =
                KVConfigSerializeWrapper.fromJson(content, KVConfigSerializeWrapper.class);
            if (null != kvConfigSerializeWrapper) { 
                // 将内容拷贝到配置表中
                this.configTable.putAll(kvConfigSerializeWrapper.getConfigTable());
                log.info("load KV config table OK");
            }
        }
    }

>  加载`kvConfig.json`文件，然后使用`FastJson`将文件内容格式化为`KVConfigSerializeWrapper`对象，最后将`KVConfigSerializeWrapper`对象中的`configTable`集合内容拷贝保存到`KVConfigManager`组件中。

### 3.2.1 获取`kvConfig.json`文件内容 ###

public static String file2String(final String fileName) throws IOException { 
        File file = new File(fileName);
        return file2String(file);
    }
    
    public static String file2String(final File file) throws IOException { 
        if (file.exists()) { 
            byte[] data = new byte[(int) file.length()];
            boolean result;
    
            FileInputStream inputStream = null;
            try { 
                inputStream = new FileInputStream(file);
                int len = inputStream.read(data);
                result = len == data.length;
            } finally { 
                if (inputStream != null) { 
                    inputStream.close();
                }
            }
    
            if (result) { 
                return new String(data);
            }
        }
        return null;
    }

> 使用文件输入流读取`kvConfig.json`的内容

### 3.2.2 `KVConfigSerializeWrapper` ###

public class KVConfigSerializeWrapper extends RemotingSerializable { 
        // 由这个HashMap集合，我们完全可以推断出kvConfig.json文件的格式
        private HashMap<String/* Namespace */, HashMap<String/* Key */, String/* Value */>> configTable;
    
        // getter方法
        public HashMap<String, HashMap<String, String>> getConfigTable() { 
            return configTable;
        }
    
        // setter方法
        public void setConfigTable(HashMap<String, HashMap<String, String>> configTable) { 
            this.configTable = configTable;
        }
    }

> 这个类是`json`字符串转换的目标对象
> 
> 借助`FastJson`进行转换

## 3.3 `printAllPeriodically()`方法，打印`configTable`的内容 ##

public void printAllPeriodically() { 
        try { 
            this.lock.readLock().lockInterruptibly();
            try { 
                log.info("--------------------------------------------------------");
    
                { 
                    log.info("configTable SIZE: {}", this.configTable.size());
                    Iterator<Entry<String, HashMap<String, String>>> it =
                        this.configTable.entrySet().iterator();
                    while (it.hasNext()) { 
                        Entry<String, HashMap<String, String>> next = it.next();
                        Iterator<Entry<String, String>> itSub = next.getValue().entrySet().iterator();
                        while (itSub.hasNext()) { 
                            Entry<String, String> nextSub = itSub.next();
                            log.info("configTable NS: {} Key: {} Value: {}", next.getKey(), nextSub.getKey(),
                                nextSub.getValue());
                        }
                    }
                }
            } finally { 
                this.lock.readLock().unlock();
            }
        } catch (InterruptedException e) { 
            log.error("printAllPeriodically InterruptedException", e);
        }
    }

> 该方法只是将`configTable`这个`map`集合的内容日志输出一下
> 
> `configTable`的内容来源见`3.2`

# 4 `RouteInfoManager`组件 #

用途：这个组件非常重要，它维护所有`Broker`的路由信息，该类中包含`5`个`HashMap`集合，如下所示

private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
    private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
    private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
    private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
    private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

>  *  `brokerLiveTable`：`key`是`Broker`节点的名字，`value`是该节点上次上报（心跳）的信息`BrokerLiveInfo`，它里面有`4`个属性
>  *  `lastUpdateTimestamp` `Broker`最后一次上报时间
>  *  `dataVersion` 数据版本
>  *  `channel` `netty`连接的通道
>  *  `haServerAddr`

## 4.1 构造方法 ##

public RouteInfoManager() { 
        this.topicQueueTable = new HashMap<String, List<QueueData>>(1024);
        this.brokerAddrTable = new HashMap<String, BrokerData>(128);
        this.clusterAddrTable = new HashMap<String, Set<String>>(32);
        this.brokerLiveTable = new HashMap<String, BrokerLiveInfo>(256);
        this.filterServerTable = new HashMap<String, List<String>>(256);
    }

> 构造这`5`个`HashMap`集合

## 4.2 `scanNotActiveBroker()`方法，扫描并剔除不活跃的`Broker`路由信息 ##

// 120秒
    private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
    
    public void scanNotActiveBroker() { 
        Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
        while (it.hasNext()) { 
            Entry<String, BrokerLiveInfo> next = it.next();
            // 获取Broker节点最后一次上报时间
            long last = next.getValue().getLastUpdateTimestamp();
            // 如果某个Broker的最后一次上报时间加上120秒，仍然小于当前系统时间
            // 那么就剔除这个Broker节点信息
            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);
                // 还要剔除其他路由表中该节点的信息，见4.3
                this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
            }
        }
    }

>  从`brokerLiveTable`这个路由表判断`Broker`节点是否按时上报，未按时上报就认为该`Broker`节点宕机，需要剔除所有路由表中宕机节点的信息。

## 4.3 `onChannelDestroy()`方法，剔除其他路由表中宕机节点信息 ##

private final ReadWriteLock lock = new ReentrantReadWriteLock();
    
    public void onChannelDestroy(String remoteAddr, Channel channel) { 
        String brokerAddrFound = null;
        if (channel != null) { 
            try { 
                try { 
                    this.lock.readLock().lockInterruptibly();
                    Iterator<Entry<String, BrokerLiveInfo>> itBrokerLiveTable =
                        this.brokerLiveTable.entrySet().iterator();
                    while (itBrokerLiveTable.hasNext()) { 
                        Entry<String, BrokerLiveInfo> entry = itBrokerLiveTable.next();
                        if (entry.getValue().getChannel() == channel) { 
                            brokerAddrFound = entry.getKey();
                            break;
                        }
                    }
                } finally { 
                    this.lock.readLock().unlock();
                }
            } catch (Exception e) { 
                log.error("onChannelDestroy Exception", e);
            }
        }
    
        if (null == brokerAddrFound) { 
            brokerAddrFound = remoteAddr;
        } else { 
            log.info("the broker's channel destroyed, {}, clean it's data structure at once", brokerAddrFound);
        }
    
        if (brokerAddrFound != null && brokerAddrFound.length() > 0) { 
    
            try { 
                try { 
                    this.lock.writeLock().lockInterruptibly();
                    this.brokerLiveTable.remove(brokerAddrFound);
                    this.filterServerTable.remove(brokerAddrFound);
                    String brokerNameFound = null;
                    boolean removeBrokerName = false;
                    Iterator<Entry<String, BrokerData>> itBrokerAddrTable =
                        this.brokerAddrTable.entrySet().iterator();
                    while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) { 
                        BrokerData brokerData = itBrokerAddrTable.next().getValue();
    
                        Iterator<Entry<Long, String>> it = brokerData.getBrokerAddrs().entrySet().iterator();
                        while (it.hasNext()) { 
                            Entry<Long, String> entry = it.next();
                            Long brokerId = entry.getKey();
                            String brokerAddr = entry.getValue();
                            if (brokerAddr.equals(brokerAddrFound)) { 
                                brokerNameFound = brokerData.getBrokerName();
                                it.remove();
                                log.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",
                                         brokerId, brokerAddr);
                                break;
                            }
                        }
    
                        if (brokerData.getBrokerAddrs().isEmpty()) { 
                            removeBrokerName = true;
                            itBrokerAddrTable.remove();
                            log.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",
                                     brokerData.getBrokerName());
                        }
                    }
    
                    if (brokerNameFound != null && removeBrokerName) { 
                        Iterator<Entry<String, Set<String>>> it = this.clusterAddrTable.entrySet().iterator();
                        while (it.hasNext()) { 
                            Entry<String, Set<String>> entry = it.next();
                            String clusterName = entry.getKey();
                            Set<String> brokerNames = entry.getValue();
                            boolean removed = brokerNames.remove(brokerNameFound);
                            if (removed) { 
                                log.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",
                                         brokerNameFound, clusterName);
    
                                if (brokerNames.isEmpty()) { 
                                    log.info("remove the clusterName[{}] from clusterAddrTable, because channel destroyed and no broker in this cluster",
                                             clusterName);
                                    it.remove();
                                }
    
                                break;
                            }
                        }
                    }
    
                    if (removeBrokerName) { 
                        Iterator<Entry<String, List<QueueData>>> itTopicQueueTable =
                            this.topicQueueTable.entrySet().iterator();
                        while (itTopicQueueTable.hasNext()) { 
                            Entry<String, List<QueueData>> entry = itTopicQueueTable.next();
                            String topic = entry.getKey();
                            List<QueueData> queueDataList = entry.getValue();
    
                            Iterator<QueueData> itQueueData = queueDataList.iterator();
                            while (itQueueData.hasNext()) { 
                                QueueData queueData = itQueueData.next();
                                if (queueData.getBrokerName().equals(brokerNameFound)) { 
                                    itQueueData.remove();
                                    log.info("remove topic[{} {}], from topicQueueTable, because channel destroyed",
                                             topic, queueData);
                                }
                            }
    
                            if (queueDataList.isEmpty()) { 
                                itTopicQueueTable.remove();
                                log.info("remove topic[{}] all queue, from topicQueueTable, because channel destroyed",
                                         topic);
                            }
                        }
                    }
                } finally { 
                    this.lock.writeLock().unlock();
                }
            } catch (Exception e) { 
                log.error("onChannelDestroy Exception", e);
            }
        }
    }

>  如果某个`Broker`节点超过`120`秒仍然没有上报它的信息，那么`Namesrv`就认为这个`Broker`节点已经宕机了，就从路由表中剔除这个`Broker`节点的信息。注意，有好几个路由表哦，都需要删除。

# 5 `BrokerHousekeepingService`组件 #

该类实现接口`ChannelEventListener`，从名字也能看出，这个类应该是一个监听器，下面是接口的方法

public interface ChannelEventListener { 
    
        // 通道连接事件
        void onChannelConnect(final String remoteAddr, final Channel channel);
    
        // 通道关闭事件
        void onChannelClose(final String remoteAddr, final Channel channel);
    
        // 通道异常事件
        void onChannelException(final String remoteAddr, final Channel channel);
    
        //
        void onChannelIdle(final String remoteAddr, final Channel channel);
    }

而其实现类中这`4`个方法的实现如下所示

public class BrokerHousekeepingService implements ChannelEventListener { 
        private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
        private final NamesrvController namesrvController;
    
        public BrokerHousekeepingService(NamesrvController namesrvController) { 
            this.namesrvController = namesrvController;
        }
    
        @Override
        public void onChannelConnect(String remoteAddr, Channel channel) { 
        }
    
        @Override
        public void onChannelClose(String remoteAddr, Channel channel) { 
            this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
        }
    
        @Override
        public void onChannelException(String remoteAddr, Channel channel) { 
            this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
        }
    
        @Override
        public void onChannelIdle(String remoteAddr, Channel channel) { 
            this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
        }
    }

>  可以发现，除了`onChannelConnect`事件不做处理之外，另外`3`个事件对应的操作都是调用`RouteInfoManager`组件的`onChannelDestroy()`方法，从路由表中剔除该连接信息，见`4.3`

>  到目前为止，`Namesrv`的启动过程就讲完了，虽然上面还有很多地方没有弄懂，但是我后面会继续去深挖它的源码，争取搞清楚它各个组件的作用以及他们是如何交互的。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMDEyOTgx_size_16_color_FFFFFF_t_70_pic_center]: /images/20221014/35bfde79758c4418a2670e08fde88f45.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMDEyOTgx_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221014/9b6280f63f674a2e98c3dde6ee34d702.png