RocketMQ（三）：broker启动逻辑

 // org.apache.rocketmq.broker.BrokerStartup#main    public static void main(String[] args) {        start(createBrokerController(args));    }    // 整个启动，基本就是创建一个 BrokerController, 然后调用其 start() 方法，所以整体实现是委托给 BrokerController 实现的。    public static BrokerController start(BrokerController controller) {        try {
            controller.start();
            String tip = "The broker[" + controller.getBrokerConfig().getBrokerName() + ", "                + controller.getBrokerAddr() + "] boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
            if (null != controller.getBrokerConfig().getNamesrvAddr()) {                tip += " and name server is " + controller.getBrokerConfig().getNamesrvAddr();            }
            log.info(tip);            System.out.printf("%s%n", tip);            return controller;        } catch (Throwable e) {            e.printStackTrace();            System.exit(-1);        }
        return null;    }    // 在看 BrokerController 实现前，我们需要是它是如何被创建的    public static BrokerController createBrokerController(String[] args) {        System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
        if (null == System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_SNDBUF_SIZE)) {            NettySystemConfig.socketSndbufSize = 131072;        }
        if (null == System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_RCVBUF_SIZE)) {            NettySystemConfig.socketRcvbufSize = 131072;        }
        try {            //PackageConflictDetect.detectFastjson();            // 创建默认配置项， -h, -n            Options options = ServerUtil.buildCommandlineOptions(new Options());            // 添加 -c, -p, -m 支持，由 PosixParser 解析变量到 commandLine 中            commandLine = ServerUtil.parseCmdLine("mqbroker", args, buildCommandlineOptions(options),                new PosixParser());            if (null == commandLine) {                System.exit(-1);            }
            final BrokerConfig brokerConfig = new BrokerConfig();            final NettyServerConfig nettyServerConfig = new NettyServerConfig();            final NettyClientConfig nettyClientConfig = new NettyClientConfig();
            nettyClientConfig.setUseTLS(Boolean.parseBoolean(System.getProperty(TLS_ENABLE,                String.valueOf(TlsSystemConfig.tlsMode == TlsMode.ENFORCING))));            // 服务端端口号默认为 10911            nettyServerConfig.setListenPort(10911);            // 初始化存储服务配置，如磁盘保存位置、持久化磁盘频率            final MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
            if (BrokerRole.SLAVE == messageStoreConfig.getBrokerRole()) {                int ratio = messageStoreConfig.getAccessMessageInMemoryMaxRatio() - 10;                messageStoreConfig.setAccessMessageInMemoryMaxRatio(ratio);            }            // 如果自定义了配置文件，则用配置文件的配置覆盖如上默认配置            if (commandLine.hasOption('c')) {                String file = commandLine.getOptionValue('c');                if (file != null) {                    configFile = file;                    InputStream in = new BufferedInputStream(new FileInputStream(file));                    properties = new Properties();                    properties.load(in);
                    properties2SystemEnv(properties);                    MixAll.properties2Object(properties, brokerConfig);                    MixAll.properties2Object(properties, nettyServerConfig);                    MixAll.properties2Object(properties, nettyClientConfig);                    MixAll.properties2Object(properties, messageStoreConfig);
                    BrokerPathConfigHelper.setBrokerConfigPath(file);                    in.close();                }            }
            MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), brokerConfig);
            if (null == brokerConfig.getRocketmqHome()) {                System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation", MixAll.ROCKETMQ_HOME_ENV);                System.exit(-2);            }
            // 验证 namesrvAddr 地址的有效性            String namesrvAddr = brokerConfig.getNamesrvAddr();            if (null != namesrvAddr) {                try {                    String[] addrArray = namesrvAddr.split(";");                    for (String addr : addrArray) {                        RemotingUtil.string2SocketAddress(addr);                    }                } catch (Exception e) {                    System.out.printf(                        "The Name Server Address[%s] illegal, please set it as follows, \"127.0.0.1:9876;192.168.0.1:9876\"%n",                        namesrvAddr);                    System.exit(-3);                }            }
            // brokerRole 默认是 ASYNC_MASTER            switch (messageStoreConfig.getBrokerRole()) {                case ASYNC_MASTER:                case SYNC_MASTER:                    brokerConfig.setBrokerId(MixAll.MASTER_ID);                    break;                case SLAVE:                    if (brokerConfig.getBrokerId() <= 0) {                        System.out.printf("Slave's brokerId must be > 0");                        System.exit(-3);                    }
                    break;                default:                    break;            }            // haPort 在普通服务监听端口上+1后得到            messageStoreConfig.setHaListenPort(nettyServerConfig.getListenPort() + 1);            // 日志配置用于打印            LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();            JoranConfigurator configurator = new JoranConfigurator();            configurator.setContext(lc);            lc.reset();            configurator.doConfigure(brokerConfig.getRocketmqHome() + "/conf/logback_broker.xml");
            if (commandLine.hasOption('p')) {                InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.BROKER_CONSOLE_NAME);                MixAll.printObjectProperties(console, brokerConfig);                MixAll.printObjectProperties(console, nettyServerConfig);                MixAll.printObjectProperties(console, nettyClientConfig);                MixAll.printObjectProperties(console, messageStoreConfig);                System.exit(0);            } else if (commandLine.hasOption('m')) {                InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.BROKER_CONSOLE_NAME);                MixAll.printObjectProperties(console, brokerConfig, true);                MixAll.printObjectProperties(console, nettyServerConfig, true);                MixAll.printObjectProperties(console, nettyClientConfig, true);                MixAll.printObjectProperties(console, messageStoreConfig, true);                System.exit(0);            }
            log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);            MixAll.printObjectProperties(log, brokerConfig);            MixAll.printObjectProperties(log, nettyServerConfig);            MixAll.printObjectProperties(log, nettyClientConfig);            MixAll.printObjectProperties(log, messageStoreConfig);            // 创建 BrokerController 实例            final BrokerController controller = new BrokerController(                brokerConfig,                nettyServerConfig,                nettyClientConfig,                messageStoreConfig);            // remember all configs to prevent discard            // 把所有获得的属性信息全部保存在 allConfigs 中            controller.getConfiguration().registerConfig(properties);            // 初始化，重要动作            boolean initResult = controller.initialize();            if (!initResult) {                controller.shutdown();                System.exit(-3);            }            // 注册关闭钩子，做优雅停机处理            Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {                private volatile boolean hasShutdown = false;                private AtomicInteger shutdownTimes = new AtomicInteger(0);
                @Override                public void run() {                    synchronized (this) {                        log.info("Shutdown hook was invoked, {}", this.shutdownTimes.incrementAndGet());                        if (!this.hasShutdown) {                            this.hasShutdown = true;                            long beginTime = System.currentTimeMillis();                            controller.shutdown();                            long consumingTimeTotal = System.currentTimeMillis() - beginTime;                            log.info("Shutdown hook over, consuming total time(ms): {}", consumingTimeTotal);                        }                    }                }            }, "ShutdownHook"));
            return controller;        } catch (Throwable e) {            e.printStackTrace();            System.exit(-1);        }
        return null;    }

public BrokerController(        final BrokerConfig brokerConfig,        final NettyServerConfig nettyServerConfig,        final NettyClientConfig nettyClientConfig,        final MessageStoreConfig messageStoreConfig    ) {        this.brokerConfig = brokerConfig;        this.nettyServerConfig = nettyServerConfig;        this.nettyClientConfig = nettyClientConfig;        this.messageStoreConfig = messageStoreConfig;        // 初始化offset管理器，主要用于处理客户端提交上来的offset信息，主要使用一个 new ConcurrentHashMap> 来保存        // 持久化时通过 encode() 后统一持久化到文件中        this.consumerOffsetManager = new ConsumerOffsetManager(this);        this.topicConfigManager = new TopicConfigManager(this);        // 接到pull请求后，先经过该处理器处理        this.pullMessageProcessor = new PullMessageProcessor(this);        // pullRequestHoldService 将被 messageArrivingListener 调用        this.pullRequestHoldService = new PullRequestHoldService(this);        this.messageArrivingListener = new NotifyMessageArrivingListener(this.pullRequestHoldService);        // consumerIdsChangeListener 将被 consumerManager 调用        // consumerManager 主要负责消费者的连接管理        this.consumerIdsChangeListener = new DefaultConsumerIdsChangeListener(this);        this.consumerManager = new ConsumerManager(this.consumerIdsChangeListener);        // BloomFilter        this.consumerFilterManager = new ConsumerFilterManager(this);        this.producerManager = new ProducerManager();        this.clientHousekeepingService = new ClientHousekeepingService(this);        this.broker2Client = new Broker2Client(this);        this.subscriptionGroupManager = new SubscriptionGroupManager(this);        // 客户端接入服务        this.brokerOuterAPI = new BrokerOuterAPI(nettyClientConfig);        this.filterServerManager = new FilterServerManager(this);        // 如果是 SLAVE 节点，则会定期从master上拉取元数据信息，topic,offset,delayOffset,subscriptionGroup        this.slaveSynchronize = new SlaveSynchronize(this);
        // 各种线程池队列配置        // 发送线程池队列，默认 10000        this.sendThreadPoolQueue = new LinkedBlockingQueue(this.brokerConfig.getSendThreadPoolQueueCapacity());        // pull线程池队列，默认 100000        this.pullThreadPoolQueue = new LinkedBlockingQueue(this.brokerConfig.getPullThreadPoolQueueCapacity());        // 查询线程池队列，默认 20000        this.queryThreadPoolQueue = new LinkedBlockingQueue(this.brokerConfig.getQueryThreadPoolQueueCapacity());        // 客户端连接管理线程池队列，默认 1000000        this.clientManagerThreadPoolQueue = new LinkedBlockingQueue(this.brokerConfig.getClientManagerThreadPoolQueueCapacity());        // 消费者管理线程池队列，默认 1000000        this.consumerManagerThreadPoolQueue = new LinkedBlockingQueue(this.brokerConfig.getConsumerManagerThreadPoolQueueCapacity());        // 心跳管理线程池队列，默认 50000        this.heartbeatThreadPoolQueue = new LinkedBlockingQueue(this.brokerConfig.getHeartbeatThreadPoolQueueCapacity());        // 事务管理器线程池队列，默认 100000        this.endTransactionThreadPoolQueue = new LinkedBlockingQueue(this.brokerConfig.getEndTransactionPoolQueueCapacity());        // 各种处理指标管理器        this.brokerStatsManager = new BrokerStatsManager(this.brokerConfig.getBrokerClusterName());        this.setStoreHost(new InetSocketAddress(this.getBrokerConfig().getBrokerIP1(), this.getNettyServerConfig().getListenPort()));
        this.brokerFastFailure = new BrokerFastFailure(this);        this.configuration = new Configuration(            log,            BrokerPathConfigHelper.getBrokerConfigPath(),            this.brokerConfig, this.nettyServerConfig, this.nettyClientConfig, this.messageStoreConfig        );    }

// org.apache.rocketmq.broker.BrokerController#initialize    public boolean initialize() throws CloneNotSupportedException {        // 初始化 topicConfigManager, 从文件中加载中出来，这里要说的是大部分的配置管理工作流程基本类似        // 因为被抽象出模板方法类 ConfigManager, 统一管理加载流程，稍后我们可以看到, ${user.home}\store\config\topics.json        boolean result = this.topicConfigManager.load();        // 加载消费偏移， ${user.home}\store\config\consumerOffset.json        result = result && this.consumerOffsetManager.load();        // 加载消费组信息， ${user.home}\store\config\subscriptionGroup.json        result = result && this.subscriptionGroupManager.load();        // 加载过滤器配置, ${user.home}\store\config\consumerFilter.json        result = result && this.consumerFilterManager.load();
        if (result) {            try {                // 加载存储服务，默认情况使用 CommitLog                 this.messageStore =                    new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener,                        this.brokerConfig);                if (messageStoreConfig.isEnableDLegerCommitLog()) {                    DLedgerRoleChangeHandler roleChangeHandler = new DLedgerRoleChangeHandler(this, (DefaultMessageStore) messageStore);                    ((DLedgerCommitLog)((DefaultMessageStore) messageStore).getCommitLog()).getdLedgerServer().getdLedgerLeaderElector().addRoleChangeHandler(roleChangeHandler);                }                this.brokerStats = new BrokerStats((DefaultMessageStore) this.messageStore);                //load plugin                // 以装饰者模式的形式将 plugin 一个个嵌入到 messageStore 中                MessageStorePluginContext context = new MessageStorePluginContext(messageStoreConfig, brokerStatsManager, messageArrivingListener, brokerConfig);                this.messageStore = MessageStoreFactory.build(context, this.messageStore);                // 添加一个前置的 CommitLogDispatcherCalcBitMap                this.messageStore.getDispatcherList().addFirst(new CommitLogDispatcherCalcBitMap(this.brokerConfig, this.consumerFilterManager));            } catch (IOException e) {                result = false;                log.error("Failed to initialize", e);            }        }        // messageStore 加载，重要        result = result && this.messageStore.load();
        if (result) {            // 初始化 ServerBootstrap            this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.clientHousekeepingService);            NettyServerConfig fastConfig = (NettyServerConfig) this.nettyServerConfig.clone();            fastConfig.setListenPort(nettyServerConfig.getListenPort() - 2);            this.fastRemotingServer = new NettyRemotingServer(fastConfig, this.clientHousekeepingService);            // 各个 executors            this.sendMessageExecutor = new BrokerFixedThreadPoolExecutor(                this.brokerConfig.getSendMessageThreadPoolNums(),                this.brokerConfig.getSendMessageThreadPoolNums(),                1000 * 60,                TimeUnit.MILLISECONDS,                this.sendThreadPoolQueue,                new ThreadFactoryImpl("SendMessageThread_"));
            this.pullMessageExecutor = new BrokerFixedThreadPoolExecutor(                this.brokerConfig.getPullMessageThreadPoolNums(),                this.brokerConfig.getPullMessageThreadPoolNums(),                1000 * 60,                TimeUnit.MILLISECONDS,                this.pullThreadPoolQueue,                new ThreadFactoryImpl("PullMessageThread_"));
            this.queryMessageExecutor = new BrokerFixedThreadPoolExecutor(                this.brokerConfig.getQueryMessageThreadPoolNums(),                this.brokerConfig.getQueryMessageThreadPoolNums(),                1000 * 60,                TimeUnit.MILLISECONDS,                this.queryThreadPoolQueue,                new ThreadFactoryImpl("QueryMessageThread_"));
            this.adminBrokerExecutor =                Executors.newFixedThreadPool(this.brokerConfig.getAdminBrokerThreadPoolNums(), new ThreadFactoryImpl(                    "AdminBrokerThread_"));
            this.clientManageExecutor = new ThreadPoolExecutor(                this.brokerConfig.getClientManageThreadPoolNums(),                this.brokerConfig.getClientManageThreadPoolNums(),                1000 * 60,                TimeUnit.MILLISECONDS,                this.clientManagerThreadPoolQueue,                new ThreadFactoryImpl("ClientManageThread_"));
            this.heartbeatExecutor = new BrokerFixedThreadPoolExecutor(                this.brokerConfig.getHeartbeatThreadPoolNums(),                this.brokerConfig.getHeartbeatThreadPoolNums(),                1000 * 60,                TimeUnit.MILLISECONDS,                this.heartbeatThreadPoolQueue,                new ThreadFactoryImpl("HeartbeatThread_", true));
            this.endTransactionExecutor = new BrokerFixedThreadPoolExecutor(                this.brokerConfig.getEndTransactionThreadPoolNums(),                this.brokerConfig.getEndTransactionThreadPoolNums(),                1000 * 60,                TimeUnit.MILLISECONDS,                this.endTransactionThreadPoolQueue,                new ThreadFactoryImpl("EndTransactionThread_"));
            this.consumerManageExecutor =                Executors.newFixedThreadPool(this.brokerConfig.getConsumerManageThreadPoolNums(), new ThreadFactoryImpl(                    "ConsumerManageThread_"));
            this.registerProcessor();
            final long initialDelay = UtilAll.computeNextMorningTimeMillis() - System.currentTimeMillis();            final long period = 1000 * 60 * 60 * 24;            // 定时任务            // 定时记录broker状态, 默认 86400s 调度一次            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {                @Override                public void run() {                    try {                        BrokerController.this.getBrokerStats().record();                    } catch (Throwable e) {                        log.error("schedule record error.", e);                    }                }            }, initialDelay, period, TimeUnit.MILLISECONDS);            // 定时持久化 offset, 5s 持久化一次            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {                @Override                public void run() {                    try {                        BrokerController.this.consumerOffsetManager.persist();                    } catch (Throwable e) {                        log.error("schedule persist consumerOffset error.", e);                    }                }            }, 1000 * 10, this.brokerConfig.getFlushConsumerOffsetInterval(), TimeUnit.MILLISECONDS);            // 定时持久化filter, 10s同步一次            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {                @Override                public void run() {                    try {                        BrokerController.this.consumerFilterManager.persist();                    } catch (Throwable e) {                        log.error("schedule persist consumer filter error.", e);                    }                }            }, 1000 * 10, 1000 * 10, TimeUnit.MILLISECONDS);            // 对处理缓慢的 broker 作保持操作，第3分钟运行一次            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {                @Override                public void run() {                    try {                        BrokerController.this.protectBroker();                    } catch (Throwable e) {                        log.error("protectBroker error.", e);                    }                }            }, 3, 3, TimeUnit.MINUTES);            // 打印waterMark, 1s 每次            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {                @Override                public void run() {                    try {                        BrokerController.this.printWaterMark();                    } catch (Throwable e) {                        log.error("printWaterMark error.", e);                    }                }            }, 10, 1, TimeUnit.SECONDS);            //             this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
                @Override                public void run() {                    try {                        log.info("dispatch behind commit log {} bytes", BrokerController.this.getMessageStore().dispatchBehindBytes());                    } catch (Throwable e) {                        log.error("schedule dispatchBehindBytes error.", e);                    }                }            }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
            if (this.brokerConfig.getNamesrvAddr() != null) {                // 从 namesrvAddr 刷新namesrvAddr                this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr());                log.info("Set user specified name server address: {}", this.brokerConfig.getNamesrvAddr());            } else if (this.brokerConfig.isFetchNamesrvAddrByAddressServer()) {                this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
                    @Override                    public void run() {                        try {                            BrokerController.this.brokerOuterAPI.fetchNameServerAddr();                        } catch (Throwable e) {                            log.error("ScheduledTask fetchNameServerAddr exception", e);                        }                    }                }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);            }
            if (!messageStoreConfig.isEnableDLegerCommitLog()) {                if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {                    if (this.messageStoreConfig.getHaMasterAddress() != null && this.messageStoreConfig.getHaMasterAddress().length() >= 6) {                        this.messageStore.updateHaMasterAddress(this.messageStoreConfig.getHaMasterAddress());                        this.updateMasterHAServerAddrPeriodically = false;                    } else {                        this.updateMasterHAServerAddrPeriodically = true;                    }                } else {                    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {                        @Override                        public void run() {                            try {                                BrokerController.this.printMasterAndSlaveDiff();                            } catch (Throwable e) {                                log.error("schedule printMasterAndSlaveDiff error.", e);                            }                        }                    }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);                }            }
            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();                                ((NettyRemotingServer) fastRemotingServer).loadSslContext();                            }                        });                } catch (Exception e) {                    log.warn("FileWatchService created error, can't load the certificate dynamically");                }            }            // 初始化事务管理器            initialTransaction();            // 权限管理            initialAcl();            // rpc钩子            initialRpcHooks();        }        return result;    }

    // ConfigManager, 抽象出的一系列模板方法，用于加载文件，持久化数据等待    // org.apache.rocketmq.common.ConfigManager#load    public boolean load() {        String fileName = null;        try {            // 由子类实现 configFilePath 的寻找，然后解出文件信息            fileName = this.configFilePath();            String jsonString = MixAll.file2String(fileName);            // 如果原来就不存在文件，则尝试使用备份的配置加载一遍(*.bak)，如果也不存在，就算了            if (null == jsonString || jsonString.length() == 0) {                return this.loadBak();            } else {                // 如果找到相应配置文件，则进行解码得到配置信息                this.decode(jsonString);                log.info("load " + fileName + " OK");                return true;            }        } catch (Exception e) {            log.error("load " + fileName + " failed, and try to load backup file", e);            return this.loadBak();        }    }    private boolean loadBak() {        String fileName = null;        try {            fileName = this.configFilePath();            String jsonString = MixAll.file2String(fileName + ".bak");            if (jsonString != null && jsonString.length() > 0) {                this.decode(jsonString);                log.info("load " + fileName + " OK");                return true;            }        } catch (Exception e) {            log.error("load " + fileName + " Failed", e);            return false;        }
        return true;    }
    // 核心存在服务构造方法    // org.apache.rocketmq.store.DefaultMessageStore#DefaultMessageStore    public DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,        final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig) throws IOException {        this.messageArrivingListener = messageArrivingListener;        this.brokerConfig = brokerConfig;        this.messageStoreConfig = messageStoreConfig;        this.brokerStatsManager = brokerStatsManager;        // 文件分配管理服务        this.allocateMappedFileService = new AllocateMappedFileService(this);        if (messageStoreConfig.isEnableDLegerCommitLog()) {            this.commitLog = new DLedgerCommitLog(this);        } else {            // 默认核心存储            this.commitLog = new CommitLog(this);        }        this.consumeQueueTable = new ConcurrentHashMap<>(32);
        this.flushConsumeQueueService = new FlushConsumeQueueService();        // 清理磁盘文件,主要磁盘占用太高时清理        this.cleanCommitLogService = new CleanCommitLogService();        // 清理过期数据文件        this.cleanConsumeQueueService = new CleanConsumeQueueService();        this.storeStatsService = new StoreStatsService();        // 索引服务        this.indexService = new IndexService(this);        if (!messageStoreConfig.isEnableDLegerCommitLog()) {            // 做高可用，主要为了保证主从同步            this.haService = new HAService(this);        } else {            this.haService = null;        }        //         this.reputMessageService = new ReputMessageService();        // 后台服务调用管理        this.scheduleMessageService = new ScheduleMessageService(this);
        this.transientStorePool = new TransientStorePool(messageStoreConfig);
        if (messageStoreConfig.isTransientStorePoolEnable()) {            this.transientStorePool.init();        }        // 开启文件预分配服务线程        this.allocateMappedFileService.start();        // 索引服务开启，默认无操作        this.indexService.start();
        this.dispatcherList = new LinkedList<>();        this.dispatcherList.addLast(new CommitLogDispatcherBuildConsumeQueue());        this.dispatcherList.addLast(new CommitLogDispatcherBuildIndex());        // ${user.home}\store\lock        File file = new File(StorePathConfigHelper.getLockFile(messageStoreConfig.getStorePathRootDir()));        MappedFile.ensureDirOK(file.getParent());        lockFile = new RandomAccessFile(file, "rw");    }    // CommitLog 构造方法    // org.apache.rocketmq.store.CommitLog#CommitLog    public CommitLog(final DefaultMessageStore defaultMessageStore) {        this.mappedFileQueue = new MappedFileQueue(defaultMessageStore.getMessageStoreConfig().getStorePathCommitLog(),            defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog(), defaultMessageStore.getAllocateMappedFileService());        this.defaultMessageStore = defaultMessageStore;
        if (FlushDiskType.SYNC_FLUSH == defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {            this.flushCommitLogService = new GroupCommitService();        } else {            // 两个刷新磁盘服务都是内部类，方便共享队列数据，控制多久进行一次持久化操作            this.flushCommitLogService = new FlushRealTimeService();        }        // Commit 线程        this.commitLogService = new CommitRealTimeService();        // append 回调        this.appendMessageCallback = new DefaultAppendMessageCallback(defaultMessageStore.getMessageStoreConfig().getMaxMessageSize());        batchEncoderThreadLocal = new ThreadLocal() {            @Override            protected MessageExtBatchEncoder initialValue() {                return new MessageExtBatchEncoder(defaultMessageStore.getMessageStoreConfig().getMaxMessageSize());            }        };        // 锁的使用方式，默认是 自旋锁实现        this.putMessageLock = defaultMessageStore.getMessageStoreConfig().isUseReentrantLockWhenPutMessage() ? new PutMessageReentrantLock() : new PutMessageSpinLock();
    }

    // 文件存储加载    // org.apache.rocketmq.store.DefaultMessageStore#load    public boolean load() {        boolean result = true;
        try {            boolean lastExitOK = !this.isTempFileExist();            log.info("last shutdown {}", lastExitOK ? "normally" : "abnormally");            // 加载调用服务的数据，如 delayOffset.json            if (null != scheduleMessageService) {                result = result && this.scheduleMessageService.load();            }
            // load Commit Log            result = result && this.commitLog.load();
            // load Consume Queue            result = result && this.loadConsumeQueue();
            if (result) {                // checkpoint                this.storeCheckpoint =                    new StoreCheckpoint(StorePathConfigHelper.getStoreCheckpoint(this.messageStoreConfig.getStorePathRootDir()));
                this.indexService.load(lastExitOK);
                this.recover(lastExitOK);
                log.info("load over, and the max phy offset = {}", this.getMaxPhyOffset());            }        } catch (Exception e) {            log.error("load exception", e);            result = false;        }
        if (!result) {            this.allocateMappedFileService.shutdown();        }
        return result;    }
    // org.apache.rocketmq.store.MappedFileQueue#load    public boolean load() {        // 列举store目录下所有文件        File dir = new File(this.storePath);        File[] files = dir.listFiles();        if (files != null) {            // ascending order            Arrays.sort(files);            for (File file : files) {                // 每个文件大小一定，会预分配                if (file.length() != this.mappedFileSize) {                    log.warn(file + "\t" + file.length()                        + " length not matched message store config value, please check it manually");                    return false;                }
                try {                    // 将文件映射为 MappedFile, 使用 FileChannel 进行读取                    MappedFile mappedFile = new MappedFile(file.getPath(), mappedFileSize);
                    mappedFile.setWrotePosition(this.mappedFileSize);                    mappedFile.setFlushedPosition(this.mappedFileSize);                    mappedFile.setCommittedPosition(this.mappedFileSize);                    this.mappedFiles.add(mappedFile);                    log.info("load " + file.getPath() + " OK");                } catch (IOException e) {                    log.error("load file " + file + " error", e);                    return false;                }            }        }
        return true;    }
    // org.apache.rocketmq.store.DefaultMessageStore#loadConsumeQueue    private boolean loadConsumeQueue() {        File dirLogic = new File(StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()));        File[] fileTopicList = dirLogic.listFiles();        if (fileTopicList != null) {            // 每个topic一个目录，每个queueId一个文件, 如: TopicTest/0,1,2,3,4,5,6,7            for (File fileTopic : fileTopicList) {                String topic = fileTopic.getName();
                File[] fileQueueIdList = fileTopic.listFiles();                if (fileQueueIdList != null) {                    for (File fileQueueId : fileQueueIdList) {                        int queueId;                        try {                            queueId = Integer.parseInt(fileQueueId.getName());                        } catch (NumberFormatException e) {                            continue;                        }                        ConsumeQueue logic = new ConsumeQueue(                            topic,                            queueId,                            StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),                            this.getMessageStoreConfig().getMappedFileSizeConsumeQueue(),                            this);                        this.putConsumeQueue(topic, queueId, logic);                        if (!logic.load()) {                            return false;                        }                    }                }            }        }
        log.info("load logics queue all over, OK");
        return true;    }
    // 将前面初始化好的 executors 注册到相应的表中    // org.apache.rocketmq.broker.BrokerController#registerProcessor    public void registerProcessor() {        /**         * SendMessageProcessor         */        SendMessageProcessor sendProcessor = new SendMessageProcessor(this);        sendProcessor.registerSendMessageHook(sendMessageHookList);        sendProcessor.registerConsumeMessageHook(consumeMessageHookList);
        this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor);        this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor);        this.remotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendProcessor, this.sendMessageExecutor);        this.remotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor, this.sendMessageExecutor);        this.fastRemotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor);        this.fastRemotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor);        this.fastRemotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendProcessor, this.sendMessageExecutor);        this.fastRemotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor, this.sendMessageExecutor);        /**         * PullMessageProcessor         */        this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE, this.pullMessageProcessor, this.pullMessageExecutor);        this.pullMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);
        /**         * QueryMessageProcessor         */        NettyRequestProcessor queryProcessor = new QueryMessageProcessor(this);        this.remotingServer.registerProcessor(RequestCode.QUERY_MESSAGE, queryProcessor, this.queryMessageExecutor);        this.remotingServer.registerProcessor(RequestCode.VIEW_MESSAGE_BY_ID, queryProcessor, this.queryMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.QUERY_MESSAGE, queryProcessor, this.queryMessageExecutor);        this.fastRemotingServer.registerProcessor(RequestCode.VIEW_MESSAGE_BY_ID, queryProcessor, this.queryMessageExecutor);
        /**         * ClientManageProcessor         */        ClientManageProcessor clientProcessor = new ClientManageProcessor(this);        this.remotingServer.registerProcessor(RequestCode.HEART_BEAT, clientProcessor, this.heartbeatExecutor);        this.remotingServer.registerProcessor(RequestCode.UNREGISTER_CLIENT, clientProcessor, this.clientManageExecutor);        this.remotingServer.registerProcessor(RequestCode.CHECK_CLIENT_CONFIG, clientProcessor, this.clientManageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.HEART_BEAT, clientProcessor, this.heartbeatExecutor);        this.fastRemotingServer.registerProcessor(RequestCode.UNREGISTER_CLIENT, clientProcessor, this.clientManageExecutor);        this.fastRemotingServer.registerProcessor(RequestCode.CHECK_CLIENT_CONFIG, clientProcessor, this.clientManageExecutor);
        /**         * ConsumerManageProcessor         */        ConsumerManageProcessor consumerManageProcessor = new ConsumerManageProcessor(this);        this.remotingServer.registerProcessor(RequestCode.GET_CONSUMER_LIST_BY_GROUP, consumerManageProcessor, this.consumerManageExecutor);        this.remotingServer.registerProcessor(RequestCode.UPDATE_CONSUMER_OFFSET, consumerManageProcessor, this.consumerManageExecutor);        this.remotingServer.registerProcessor(RequestCode.QUERY_CONSUMER_OFFSET, consumerManageProcessor, this.consumerManageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.GET_CONSUMER_LIST_BY_GROUP, consumerManageProcessor, this.consumerManageExecutor);        this.fastRemotingServer.registerProcessor(RequestCode.UPDATE_CONSUMER_OFFSET, consumerManageProcessor, this.consumerManageExecutor);        this.fastRemotingServer.registerProcessor(RequestCode.QUERY_CONSUMER_OFFSET, consumerManageProcessor, this.consumerManageExecutor);
        /**         * EndTransactionProcessor         */        this.remotingServer.registerProcessor(RequestCode.END_TRANSACTION, new EndTransactionProcessor(this), this.endTransactionExecutor);        this.fastRemotingServer.registerProcessor(RequestCode.END_TRANSACTION, new EndTransactionProcessor(this), this.endTransactionExecutor);
        /**         * Default         */        AdminBrokerProcessor adminProcessor = new AdminBrokerProcessor(this);        this.remotingServer.registerDefaultProcessor(adminProcessor, this.adminBrokerExecutor);        this.fastRemotingServer.registerDefaultProcessor(adminProcessor, this.adminBrokerExecutor);    }
    // org.apache.rocketmq.broker.BrokerController#initialTransaction    private void initialTransaction() {        this.transactionalMessageService = ServiceProvider.loadClass(ServiceProvider.TRANSACTION_SERVICE_ID, TransactionalMessageService.class);        if (null == this.transactionalMessageService) {            this.transactionalMessageService = new TransactionalMessageServiceImpl(new TransactionalMessageBridge(this, this.getMessageStore()));            log.warn("Load default transaction message hook service: {}", TransactionalMessageServiceImpl.class.getSimpleName());        }        this.transactionalMessageCheckListener = ServiceProvider.loadClass(ServiceProvider.TRANSACTION_LISTENER_ID, AbstractTransactionalMessageCheckListener.class);        if (null == this.transactionalMessageCheckListener) {            this.transactionalMessageCheckListener = new DefaultTransactionalMessageCheckListener();            log.warn("Load default discard message hook service: {}", DefaultTransactionalMessageCheckListener.class.getSimpleName());        }        this.transactionalMessageCheckListener.setBrokerController(this);        this.transactionalMessageCheckService = new TransactionalMessageCheckService(this);    }

// org.apache.rocketmq.broker.BrokerController#start    public void start() throws Exception {        // messageStore 服务开启        if (this.messageStore != null) {            this.messageStore.start();        }        // 10911        if (this.remotingServer != null) {            this.remotingServer.start();        }        // 10909        if (this.fastRemotingServer != null) {            this.fastRemotingServer.start();        }        // 监控文件变化，以 listener.onChanged(watchFiles.get(i)); 进行通知        if (this.fileWatchService != null) {            this.fileWatchService.start();        }        // remotingClient.start()        if (this.brokerOuterAPI != null) {            this.brokerOuterAPI.start();        }
        if (this.pullRequestHoldService != null) {            this.pullRequestHoldService.start();        }        // 清理过期连接        // brokerController.getProducerManager().scanNotActiveChannel();        // brokerController.getConsumerManager().scanNotActiveChannel();        // brokerController.getFilterServerManager().scanNotActiveChannel();        if (this.clientHousekeepingService != null) {            this.clientHousekeepingService.start();        }
        if (this.filterServerManager != null) {            this.filterServerManager.start();        }
        if (!messageStoreConfig.isEnableDLegerCommitLog()) {            startProcessorByHa(messageStoreConfig.getBrokerRole());            handleSlaveSynchronize(messageStoreConfig.getBrokerRole());        }


        this.registerBrokerAll(true, false, true);
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override            public void run() {                try {                    BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());                } catch (Throwable e) {                    log.error("registerBrokerAll Exception", e);                }            }        }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);
        if (this.brokerStatsManager != null) {            this.brokerStatsManager.start();        }
        if (this.brokerFastFailure != null) {            this.brokerFastFailure.start();        }

    }
    // 启动 messageStore 及相关的刷新服务线程    // org.apache.rocketmq.store.DefaultMessageStore#start    public void start() throws Exception {
        lock = lockFile.getChannel().tryLock(0, 1, false);        if (lock == null || lock.isShared() || !lock.isValid()) {            throw new RuntimeException("Lock failed,MQ already started");        }
        lockFile.getChannel().write(ByteBuffer.wrap("lock".getBytes()));        lockFile.getChannel().force(true);        {            /**             * 1. Make sure the fast-forward messages to be truncated during the recovering according to the max physical offset of the commitlog;             * 2. DLedger committedPos may be missing, so the maxPhysicalPosInLogicQueue maybe bigger that maxOffset returned by DLedgerCommitLog, just let it go;             * 3. Calculate the reput offset according to the consume queue;             * 4. Make sure the fall-behind messages to be dispatched before starting the commitlog, especially when the broker role are automatically changed.             */            long maxPhysicalPosInLogicQueue = commitLog.getMinOffset();            for (ConcurrentMap maps : this.consumeQueueTable.values()) {                for (ConsumeQueue logic : maps.values()) {                    if (logic.getMaxPhysicOffset() > maxPhysicalPosInLogicQueue) {                        maxPhysicalPosInLogicQueue = logic.getMaxPhysicOffset();                    }                }            }            if (maxPhysicalPosInLogicQueue < 0) {                maxPhysicalPosInLogicQueue = 0;            }            if (maxPhysicalPosInLogicQueue < this.commitLog.getMinOffset()) {                maxPhysicalPosInLogicQueue = this.commitLog.getMinOffset();                /**                 * This happens in following conditions:                 * 1. If someone removes all the consumequeue files or the disk get damaged.                 * 2. Launch a new broker, and copy the commitlog from other brokers.                 *                 * All the conditions has the same in common that the maxPhysicalPosInLogicQueue should be 0.                 * If the maxPhysicalPosInLogicQueue is gt 0, there maybe something wrong.                 */                log.warn("[TooSmallCqOffset] maxPhysicalPosInLogicQueue={} clMinOffset={}", maxPhysicalPosInLogicQueue, this.commitLog.getMinOffset());            }            log.info("[SetReputOffset] maxPhysicalPosInLogicQueue={} clMinOffset={} clMaxOffset={} clConfirmedOffset={}",                maxPhysicalPosInLogicQueue, this.commitLog.getMinOffset(), this.commitLog.getMaxOffset(), this.commitLog.getConfirmOffset());            this.reputMessageService.setReputFromOffset(maxPhysicalPosInLogicQueue);            this.reputMessageService.start();
            /**             *  1. Finish dispatching the messages fall behind, then to start other services.             *  2. DLedger committedPos may be missing, so here just require dispatchBehindBytes <= 0             */            while (true) {                if (dispatchBehindBytes() <= 0) {                    break;                }                Thread.sleep(1000);                log.info("Try to finish doing reput the messages fall behind during the starting, reputOffset={} maxOffset={} behind={}", this.reputMessageService.getReputFromOffset(), this.getMaxPhyOffset(), this.dispatchBehindBytes());            }            this.recoverTopicQueueTable();        }
        if (!messageStoreConfig.isEnableDLegerCommitLog()) {            this.haService.start();            this.handleScheduleMessageService(messageStoreConfig.getBrokerRole());        }        // 刷盘服务，commitLog服务        this.flushConsumeQueueService.start();        this.commitLog.start();        this.storeStatsService.start();
        this.createTempFile();        this.addScheduleTask();        this.shutdown = false;    }    // org.apache.rocketmq.store.DefaultMessageStore#addScheduleTask    private void addScheduleTask() {        // 清理过期数据文件        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {            @Override            public void run() {                DefaultMessageStore.this.cleanFilesPeriodically();            }        }, 1000 * 60, this.messageStoreConfig.getCleanResourceInterval(), TimeUnit.MILLISECONDS);        // 自检        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {            @Override            public void run() {                DefaultMessageStore.this.checkSelf();            }        }, 1, 10, TimeUnit.MINUTES);        //         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {            @Override            public void run() {                if (DefaultMessageStore.this.getMessageStoreConfig().isDebugLockEnable()) {                    try {                        if (DefaultMessageStore.this.commitLog.getBeginTimeInLock() != 0) {                            long lockTime = System.currentTimeMillis() - DefaultMessageStore.this.commitLog.getBeginTimeInLock();                            if (lockTime > 1000 && lockTime < 10000000) {
                                String stack = UtilAll.jstack();                                final String fileName = System.getProperty("user.home") + File.separator + "debug/lock/stack-"                                    + DefaultMessageStore.this.commitLog.getBeginTimeInLock() + "-" + lockTime;                                MixAll.string2FileNotSafe(stack, fileName);                            }                        }                    } catch (Exception e) {                    }                }            }        }, 1, 1, TimeUnit.SECONDS);
        // this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {        // @Override        // public void run() {        // DefaultMessageStore.this.cleanExpiredConsumerQueue();        // }        // }, 1, 1, TimeUnit.HOURS);    }
    // org.apache.rocketmq.remoting.netty.NettyRemotingServer#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)                .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())                .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())                // 端口绑定                .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))                .childHandler(new ChannelInitializer() {                    @Override                    public void initChannel(SocketChannel ch) throws Exception {                        // 添加一系列的 handler                        // org.apache.rocketmq.remoting.netty.NettyRemotingServer.HandshakeHandler                        // org.apache.rocketmq.remoting.netty.NettyEncoder                        // org.apache.rocketmq.remoting.netty.NettyDecoder                        // io.netty.handler.timeout.IdleStateHandler                        // org.apache.rocketmq.remoting.netty.NettyRemotingServer.NettyConnectManageHandler                        // org.apache.rocketmq.remoting.netty.NettyRemotingServer.NettyServerHandler                        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();        }        // 后台一直在扫描超时的请求（怎么都感觉不优雅）        this.timer.scheduleAtFixedRate(new TimerTask() {
            @Override            public void run() {                try {                    NettyRemotingServer.this.scanResponseTable();                } catch (Throwable e) {                    log.error("scanResponseTable exception", e);                }            }        }, 1000 * 3, 1000);    }    // 客户端连接准备，后续由 getChannel() 创建连接    // org.apache.rocketmq.remoting.netty.NettyRemotingClient#start    @Override    public void start() {        this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(            nettyClientConfig.getClientWorkerThreads(),            new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);
                @Override                public Thread newThread(Runnable r) {                    return new Thread(r, "NettyClientWorkerThread_" + this.threadIndex.incrementAndGet());                }            });
        Bootstrap handler = this.bootstrap.group(this.eventLoopGroupWorker).channel(NioSocketChannel.class)            .option(ChannelOption.TCP_NODELAY, true)            .option(ChannelOption.SO_KEEPALIVE, false)            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, nettyClientConfig.getConnectTimeoutMillis())            .option(ChannelOption.SO_SNDBUF, nettyClientConfig.getClientSocketSndBufSize())            .option(ChannelOption.SO_RCVBUF, nettyClientConfig.getClientSocketRcvBufSize())            .handler(new ChannelInitializer() {                @Override                public void initChannel(SocketChannel ch) throws Exception {                    ChannelPipeline pipeline = ch.pipeline();                    if (nettyClientConfig.isUseTLS()) {                        if (null != sslContext) {                            pipeline.addFirst(defaultEventExecutorGroup, "sslHandler", sslContext.newHandler(ch.alloc()));                            log.info("Prepend SSL handler");                        } else {                            log.warn("Connections are insecure as SSLContext is null!");                        }                    }                    pipeline.addLast(                        defaultEventExecutorGroup,                        new NettyEncoder(),                        new NettyDecoder(),                        new IdleStateHandler(0, 0, nettyClientConfig.getClientChannelMaxIdleTimeSeconds()),                        new NettyConnectManageHandler(),                        new NettyClientHandler());                }            });
        this.timer.scheduleAtFixedRate(new TimerTask() {            @Override            public void run() {                try {                    NettyRemotingClient.this.scanResponseTable();                } catch (Throwable e) {                    log.error("scanResponseTable exception", e);                }            }        }, 1000 * 3, 1000);
        if (this.channelEventListener != null) {            this.nettyEventExecutor.start();        }    }

  // org.apache.rocketmq.broker.out.BrokerOuterAPI#registerBrokerAll    public List registerBrokerAll(        final String clusterName,        final String brokerAddr,        final String brokerName,        final long brokerId,        final String haServerAddr,        final TopicConfigSerializeWrapper topicConfigWrapper,        final List filterServerList,        final boolean oneway,        final int timeoutMills,        final boolean compressed) {
        final List registerBrokerResultList = Lists.newArrayList();        List nameServerAddressList = this.remotingClient.getNameServerAddressList();        if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
            final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();            requestHeader.setBrokerAddr(brokerAddr);            requestHeader.setBrokerId(brokerId);            requestHeader.setBrokerName(brokerName);            requestHeader.setClusterName(clusterName);            requestHeader.setHaServerAddr(haServerAddr);            requestHeader.setCompressed(compressed);
            RegisterBrokerBody requestBody = new RegisterBrokerBody();            requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);            requestBody.setFilterServerList(filterServerList);            final byte[] body = requestBody.encode(compressed);            final int bodyCrc32 = UtilAll.crc32(body);            requestHeader.setBodyCrc32(bodyCrc32);            final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());            // 向所有 nameserver 注册 broker            for (final String namesrvAddr : nameServerAddressList) {                brokerOuterExecutor.execute(new Runnable() {                    @Override                    public void run() {                        try {                            // 所谓注册就是向 namesrvAddr 发送一个 请求消息                            RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);                            if (result != null) {                                registerBrokerResultList.add(result);                            }
                            log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);                        } catch (Exception e) {                            log.warn("registerBroker Exception, {}", namesrvAddr, e);                        } finally {                            countDownLatch.countDown();                        }                    }                });            }
            try {                // 略作等待，然后返回                countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);            } catch (InterruptedException e) {            }        }
        return registerBrokerResultList;    }

 // org.apache.rocketmq.broker.BrokerController#shutdown    public void shutdown() {        // broker 状态管理服务关闭        if (this.brokerStatsManager != null) {            this.brokerStatsManager.shutdown();        }        //         if (this.clientHousekeepingService != null) {            this.clientHousekeepingService.shutdown();        }        // 拉取消费服务连接管理        if (this.pullRequestHoldService != null) {            this.pullRequestHoldService.shutdown();        }        // 关闭远程服务        if (this.remotingServer != null) {            this.remotingServer.shutdown();        }
        if (this.fastRemotingServer != null) {            this.fastRemotingServer.shutdown();        }
        if (this.fileWatchService != null) {            this.fileWatchService.shutdown();        }        // 存储服务关闭        if (this.messageStore != null) {            this.messageStore.shutdown();        }        // 定时调用服务关闭        this.scheduledExecutorService.shutdown();        try {            this.scheduledExecutorService.awaitTermination(5000, TimeUnit.MILLISECONDS);        } catch (InterruptedException e) {        }        // 通知 namesrv 下线        this.unregisterBrokerAll();        // 发送线程关闭        if (this.sendMessageExecutor != null) {            this.sendMessageExecutor.shutdown();        }
        if (this.pullMessageExecutor != null) {            this.pullMessageExecutor.shutdown();        }
        if (this.adminBrokerExecutor != null) {            this.adminBrokerExecutor.shutdown();        }        // 客户端api关闭        if (this.brokerOuterAPI != null) {            this.brokerOuterAPI.shutdown();        }        // 偏移量持久化        this.consumerOffsetManager.persist();
        if (this.filterServerManager != null) {            this.filterServerManager.shutdown();        }
        if (this.brokerFastFailure != null) {            this.brokerFastFailure.shutdown();        }
        if (this.consumerFilterManager != null) {            this.consumerFilterManager.persist();        }
        if (this.clientManageExecutor != null) {            this.clientManageExecutor.shutdown();        }
        if (this.queryMessageExecutor != null) {            this.queryMessageExecutor.shutdown();        }
        if (this.consumerManageExecutor != null) {            this.consumerManageExecutor.shutdown();        }
        if (this.fileWatchService != null) {            this.fileWatchService.shutdown();        }        if (this.transactionalMessageCheckService != null) {            this.transactionalMessageCheckService.shutdown(false);        }
        if (this.endTransactionExecutor != null) {            this.endTransactionExecutor.shutdown();        }    }