RocketMQ源码解析二（Broker启动流程）

本是古典何须时尚 2022-09-12 07:46 17阅读 0赞

> **RocketMQ版本4.6.0，记录自己看源码的过程**

## 启动流程 ##

broker入口在BrokerStartup类中main()方法

public static void main(String[] args) { 
        // 1、解析并创建相关配置，根据配置创建BrokerController组件
        // 2、初始化BrokerController组件，，主要是创建一个netty服务器以及一些线程池
        // 3、启动BrokerController中的各个功能组件，开启注册和发送心跳的定时任务
        start(createBrokerController(args));
    }

嗯，看起来跟namesrv启动过程差不多。  
接下来就看上面的3个步骤  
首先是解析并创建BrokerController组件，也是有一个核心组件：

/** * 解析并创建相关配置，根据配置创建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();
            Options options = ServerUtil.buildCommandlineOptions(new Options());
            commandLine = ServerUtil.parseCmdLine("mqbroker", args, buildCommandlineOptions(options),
                new PosixParser());
            if (null == commandLine) { 
                System.exit(-1);
            }
    
            // Broker核心配置
            final BrokerConfig brokerConfig = new BrokerConfig();
            // Broker分别作为netty服务端（与生产者消费者通信）和客户端的配置（与namesrv通信）
            final NettyServerConfig nettyServerConfig = new NettyServerConfig();
            final NettyClientConfig nettyClientConfig = new NettyClientConfig();
    
            nettyClientConfig.setUseTLS(Boolean.parseBoolean(System.getProperty(TLS_ENABLE,
                String.valueOf(TlsSystemConfig.tlsMode == TlsMode.ENFORCING))));
            // 设置netty服务器监听端口10911
            nettyServerConfig.setListenPort(10911);
            // 消息存储的相关配置
            final MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
    
            // 如果是slave，就设置一个accessMessageInMemoryMaxRatio属性，暂时不懂
            if (BrokerRole.SLAVE == messageStoreConfig.getBrokerRole()) { 
                int ratio = messageStoreConfig.getAccessMessageInMemoryMaxRatio() - 10;
                messageStoreConfig.setAccessMessageInMemoryMaxRatio(ratio);
            }
    
            // 如果启动broker带了-c选项，读取-c后面的配置文件地址
            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();
                }
            }
            // 将配置解析到brokerConfig中
            MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), brokerConfig);
    
            // 没有ROCKETMQ_HOME这个环境变量，无法启动
            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);
                }
            }
    
            // 根据角色判断
            switch (messageStoreConfig.getBrokerRole()) { 
                case ASYNC_MASTER:
                case SYNC_MASTER:
                    // master就设置id为0
                    brokerConfig.setBrokerId(MixAll.MASTER_ID);
                    break;
                case SLAVE:
                    // slave的id必须大于0
                    if (brokerConfig.getBrokerId() <= 0) { 
                        System.out.print("Slave's brokerId must be > 0");
                        System.exit(-3);
                    }
    
                    break;
                default:
                    break;
            }
    
            // 如果配置了DLeger技术，则id设为-1
            if (messageStoreConfig.isEnableDLegerCommitLog()) { 
                brokerConfig.setBrokerId(-1);
            }
    
            // 设置HA监听端口
            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");
    
            // 解析启动参数中的-p和-m选项
            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);
    
            // 根据上面的4个配置创建BrokerController
            final BrokerController controller = new BrokerController(
                brokerConfig,
                nettyServerConfig,
                nettyClientConfig,
                messageStoreConfig);
            // remember all configs to prevent discard
            controller.getConfiguration().registerConfig(properties);
    
            // 初始化BrokerController组件，主要是创建一个netty服务器以及一些线程池
            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;
    }

也是有几个核心配置，BrokerConfig，NettyServerConfig，NettyClientConfig，MessageStoreConfig，其中BrokerConfig是broker自己本身的配置，NettyServerConfig是broker需要作为服务器端与生产者和消费端进行通信的配置，NettyClientConfig是broker作为客户端与namesrv通信的配置，至于里面的具体配置内容就不分析了，等具体用到了再说明。  
然后根据上面的4个配置创建BrokerController：

final BrokerController controller = new BrokerController(
        brokerConfig,
        nettyServerConfig,
        nettyClientConfig,
        messageStoreConfig);

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;
        // 创建broker各个功能对应的组件
        this.consumerOffsetManager = new ConsumerOffsetManager(this);
        this.topicConfigManager = new TopicConfigManager(this);
        this.pullMessageProcessor = new PullMessageProcessor(this);
        this.pullRequestHoldService = new PullRequestHoldService(this);
        this.messageArrivingListener = new NotifyMessageArrivingListener(this.pullRequestHoldService);
        this.consumerIdsChangeListener = new DefaultConsumerIdsChangeListener(this);
        this.consumerManager = new ConsumerManager(this.consumerIdsChangeListener);
        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);
    
        this.slaveSynchronize = new SlaveSynchronize(this);
    
        // 创建各线程池的等待队列
        this.sendThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getSendThreadPoolQueueCapacity());
        this.pullThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getPullThreadPoolQueueCapacity());
        this.replyThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getReplyThreadPoolQueueCapacity());
        this.queryThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getQueryThreadPoolQueueCapacity());
        this.clientManagerThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getClientManagerThreadPoolQueueCapacity());
        this.consumerManagerThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getConsumerManagerThreadPoolQueueCapacity());
        this.heartbeatThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getHeartbeatThreadPoolQueueCapacity());
        this.endTransactionThreadPoolQueue = new LinkedBlockingQueue<Runnable>(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
        );
    }

在BrokerController的构造方法里会同时创建各个功能组件。  
然后就需要对BrokerController进行初始化：

boolean initResult = controller.initialize();

/** * 对BrokerController进行初始化，主要是创建一个netty服务器以及一些线程池 */
    public boolean initialize() throws CloneNotSupportedException { 
        // 加载配置
        boolean result = this.topicConfigManager.load();
    
        result = result && this.consumerOffsetManager.load();
        result = result && this.subscriptionGroupManager.load();
        result = result && this.consumerFilterManager.load();
    
        if (result) { 
            try { 
                // 创建消息存储相关组件及服务
                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
                MessageStorePluginContext context = new MessageStorePluginContext(messageStoreConfig, brokerStatsManager, messageArrivingListener, brokerConfig);
                this.messageStore = MessageStoreFactory.build(context, this.messageStore);
                this.messageStore.getDispatcherList().addFirst(new CommitLogDispatcherCalcBitMap(this.brokerConfig, this.consumerFilterManager));
            } catch (IOException e) { 
                result = false;
                log.error("Failed to initialize", e);
            }
        }
    
        // 加载消息存储相关信息
        result = result && this.messageStore.load();
    
        if (result) { 
            // 创建一个netty服务器，毕竟也要生产者消费者通信
            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);
            // 下面是一些处理不同请求的线程池
            // 创建处理发送过来的消息的线程池
            this.sendMessageExecutor = new BrokerFixedThreadPoolExecutor(
                this.brokerConfig.getSendMessageThreadPoolNums(),
                this.brokerConfig.getSendMessageThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.sendThreadPoolQueue,
                new ThreadFactoryImpl("SendMessageThread_"));
            // 创建处理consumer过来拉取消息的线程池
            this.pullMessageExecutor = new BrokerFixedThreadPoolExecutor(
                this.brokerConfig.getPullMessageThreadPoolNums(),
                this.brokerConfig.getPullMessageThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.pullThreadPoolQueue,
                new ThreadFactoryImpl("PullMessageThread_"));
    
            // 回复消息的线程池
            this.replyMessageExecutor = new BrokerFixedThreadPoolExecutor(
                this.brokerConfig.getProcessReplyMessageThreadPoolNums(),
                this.brokerConfig.getProcessReplyMessageThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.replyThreadPoolQueue,
                new ThreadFactoryImpl("ProcessReplyMessageThread_"));
    
            // 查询消息的线程池
            this.queryMessageExecutor = new BrokerFixedThreadPoolExecutor(
                this.brokerConfig.getQueryMessageThreadPoolNums(),
                this.brokerConfig.getQueryMessageThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.queryThreadPoolQueue,
                new ThreadFactoryImpl("QueryMessageThread_"));
    
            // 管理broker一些命令执行的线程池
            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_"));
    
            // 管理consumer的线程池
            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统计的任务
            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);
    
            // 定时持久化消费进度
            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);
    
            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);
    
            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);
    
            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) { 
                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();
            initialRpcHooks();
        }
        return result;
    }

代码比较长，相关的流程都写在注释里了，简单概括就是加载配置和消费进度，将消费进度从磁盘读到内存，创建消息存储相关组件及服务以及加载该服务，创建一个netty服务器，毕竟也要和生产者消费者通信，创建一些处理不同请求的线程池和处理器（**这里就会有专门的线程池和处理器，不像namesrv就只有一个默认的，毕竟broker要处理的东西比较多**），以及开启一些定时任务。  
最后一步就是开启BrokerController：

controller.start();

/** * 启动各个组件 */
    public void start() throws Exception { 
        if (this.messageStore != null) { 
            // 启动存储组件里的各个后台功能，比如定时刷盘，indexFile服务等
            this.messageStore.start();
        }
    
        if (this.remotingServer != null) { 
            this.remotingServer.start();
        }
    
        if (this.fastRemotingServer != null) { 
            this.fastRemotingServer.start();
        }
    
        if (this.fileWatchService != null) { 
            this.fileWatchService.start();
        }
    
        // 这个组件就是让Broker通过Netty客户端去发送请求出去给别人
        if (this.brokerOuterAPI != null) { 
            this.brokerOuterAPI.start();
        }
    
        if (this.pullRequestHoldService != null) { 
            this.pullRequestHoldService.start();
        }
    
        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);
        }
    
        // 开一个任务去向namesrv注册，第一次是注册，后续就相当于发送心跳了
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    
            // 可以看到发送心跳间隔只能在10s-60s之间
            @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();
        }
    }

可以看到是启动一些组件，但这些组件里又会启动一些后台任务，这些等后面解析具体功能时再说明。最后还会显式开启一个定时任务，用于broker向namesrv发送心跳。

好了，最后再来总结下broker启动流程：先解析并创建4个核心配置，根据这4个配置创建BrokerController组件，然后对该组件进行初始化，在初始化的过程中会创建一些必要的功能组件，netty客户端，netty服务端，请求处理器和线程池以及开启一些定时任务，最后再启动各个功能组件并创建一个发送心跳的定时任务。  
![在这里插入图片描述][918d062dccd22a173a588550cd1184f7.png_pic_center]

## 路由注册和心跳流程 ##

这节来研究broker向namesrv注册以及发送心跳的流程（同一个定时器）

Broker启动时向集群中所有的NameServer发送心跳，每隔30s发送一次，NameServer收到Broker心跳包时会更新brokerLiveTable的lastUpdateTimestamp，然后NameServer每隔10s扫描brokerLiveTable，如果连续120s没有收到心跳包，NameServer将移除该Broker的路由信息同时关闭Socket连接。

### broker发送心跳包 ###

入口代码在启动BrokerController里

// 开一个任务去向namesrv注册，第一次是注册，后续就相当于发送心跳了
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    
        // 可以看到发送心跳间隔只能在10s-60s之间
        @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);

定时器中调用BrokerController的registerBrokerAll方法：

/** * 去向namesrv注册 */
    public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway, boolean forceRegister) { 
        TopicConfigSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildTopicConfigSerializeWrapper();
        // 不懂，跳过
        if (!PermName.isWriteable(this.getBrokerConfig().getBrokerPermission())
            || !PermName.isReadable(this.getBrokerConfig().getBrokerPermission())) { 
            ConcurrentHashMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<String, TopicConfig>();
            for (TopicConfig topicConfig : topicConfigWrapper.getTopicConfigTable().values()) { 
                TopicConfig tmp =
                    new TopicConfig(topicConfig.getTopicName(), topicConfig.getReadQueueNums(), topicConfig.getWriteQueueNums(),
                        this.brokerConfig.getBrokerPermission());
                topicConfigTable.put(topicConfig.getTopicName(), tmp);
            }
            topicConfigWrapper.setTopicConfigTable(topicConfigTable);
        }
    
        // 判断是否需要注册
        if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),
            this.getBrokerAddr(),
            this.brokerConfig.getBrokerName(),
            this.brokerConfig.getBrokerId(),
            this.brokerConfig.getRegisterBrokerTimeoutMills())) { 
            // 注册
            doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);
        }
    }

其中，forceRegister默认为true，所以到下面注册的逻辑里

private void doRegisterBrokerAll(boolean checkOrderConfig, boolean oneway,
        TopicConfigSerializeWrapper topicConfigWrapper) { 
        // 通过brokerOuterAPI组件发送网络请求到每个NameServer去注册
        List<RegisterBrokerResult> registerBrokerResultList = this.brokerOuterAPI.registerBrokerAll(
            this.brokerConfig.getBrokerClusterName(),
            this.getBrokerAddr(),
            this.brokerConfig.getBrokerName(),
            this.brokerConfig.getBrokerId(),
            this.getHAServerAddr(),
            topicConfigWrapper,
            this.filterServerManager.buildNewFilterServerList(),
            oneway,
            this.brokerConfig.getRegisterBrokerTimeoutMills(),
            this.brokerConfig.isCompressedRegister());
    
        if (registerBrokerResultList.size() > 0) { 
            // 获取第一个注册结果进行处理
            RegisterBrokerResult registerBrokerResult = registerBrokerResultList.get(0);
            if (registerBrokerResult != null) { 
                if (this.updateMasterHAServerAddrPeriodically && registerBrokerResult.getHaServerAddr() != null) { 
                    this.messageStore.updateHaMasterAddress(registerBrokerResult.getHaServerAddr());
                }
    
                this.slaveSynchronize.setMasterAddr(registerBrokerResult.getMasterAddr());
    
                if (checkOrderConfig) { 
                    this.getTopicConfigManager().updateOrderTopicConfig(registerBrokerResult.getKvTable());
                }
            }
        }
    }

通过brokerOuterAPI组件去发送心跳

/** * 构建请求去注册Broker */
    public List<RegisterBrokerResult> registerBrokerAll(
        final String clusterName,
        final String brokerAddr,
        final String brokerName,
        final long brokerId,
        final String haServerAddr,
        final TopicConfigSerializeWrapper topicConfigWrapper,
        final List<String> filterServerList,
        final boolean oneway,
        final int timeoutMills,
        final boolean compressed) { 
    
        // 初始化注册结果列表
        final List<RegisterBrokerResult> registerBrokerResultList = Lists.newArrayList();
        // nameServer地址列表
        List<String> 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);
            // 所有nameserver都注册完后再继续执行
            final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
            // 循环向每个nameserver注册
            for (final String namesrvAddr : nameServerAddressList) { 
                // 提交任务给线程池同时向每个nameserver注册
                brokerOuterExecutor.execute(new Runnable() { 
                    @Override
                    public void run() { 
                        try { 
                            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 { 
                // 都向每个nameserver注册完才继续往下，才算注册完成
                countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) { 
            }
        }
    
        return registerBrokerResultList;
    }

通过将数据封装在RegisterBrokerRequestHeader中发送请求。  
通过CountDownLatch来保证等待向所有nameserver发送完心跳后返回。

/** * 根据nameserver地址、构建的请求头和请求体去注册broker */
    private RegisterBrokerResult registerBroker(
        final String namesrvAddr,
        final boolean oneway,
        final int timeoutMills,
        final RegisterBrokerRequestHeader requestHeader,
        final byte[] body
    ) throws RemotingCommandException, MQBrokerException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
        InterruptedException { 
        // 构建一个注册的请求数据，其实请求数据和响应数据都是RemotingCommand
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.REGISTER_BROKER, requestHeader);
        request.setBody(body);
    
        // 单向请求，表示不用等待注册结果，返回null
        if (oneway) { 
            try { 
                this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);
            } catch (RemotingTooMuchRequestException e) { 
                // Ignore
            }
            return null;
        }
    
        // 同步调用
        RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);
        assert response != null;
        // 根据响应码判断注册是否成功
        switch (response.getCode()) { 
            case ResponseCode.SUCCESS: { 
                RegisterBrokerResponseHeader responseHeader =
                    (RegisterBrokerResponseHeader) response.decodeCommandCustomHeader(RegisterBrokerResponseHeader.class);
                RegisterBrokerResult result = new RegisterBrokerResult();
                result.setMasterAddr(responseHeader.getMasterAddr());
                result.setHaServerAddr(responseHeader.getHaServerAddr());
                if (response.getBody() != null) { 
                    result.setKvTable(KVTable.decode(response.getBody(), KVTable.class));
                }
                return result;
            }
            default:
                break;
        }
    
        // 失败则抛异常
        throw new MQBrokerException(response.getCode(), response.getRemark());
    }

发送心跳是同步请求，通过NettyRemotingClient发送请求，NettyRemotingClient可以看作是netty客户端，专门用来发送请求，与具体业务无关。  
**NettyRemotingClient：**

/** * netty客户端同步调用 */
    @Override
    public RemotingCommand invokeSync(String addr, final RemotingCommand request, long timeoutMillis)
        throws InterruptedException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException { 
        long beginStartTime = System.currentTimeMillis();
        // 从缓存获取或创建channel
        final Channel channel = this.getAndCreateChannel(addr);
        if (channel != null && channel.isActive()) { 
            try { 
                doBeforeRpcHooks(addr, request);
                long costTime = System.currentTimeMillis() - beginStartTime;
                // 已经超时了
                if (timeoutMillis < costTime) { 
                    throw new RemotingTimeoutException("invokeSync call timeout");
                }
                // 同步调用
                RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis - costTime);
                doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(channel), request, response);
                return response;
            } catch (RemotingSendRequestException e) { 
                log.warn("invokeSync: send request exception, so close the channel[{}]", addr);
                this.closeChannel(addr, channel);
                throw e;
            } catch (RemotingTimeoutException e) { 
                if (nettyClientConfig.isClientCloseSocketIfTimeout()) { 
                    this.closeChannel(addr, channel);
                    log.warn("invokeSync: close socket because of timeout, {}ms, {}", timeoutMillis, addr);
                }
                log.warn("invokeSync: wait response timeout exception, the channel[{}]", addr);
                throw e;
            }
        } else { 
            this.closeChannel(addr, channel);
            throw new RemotingConnectException(addr);
        }
    }
    
    public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request,
        final long timeoutMillis)
        throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException { 
        // 可以看作request id
        final int opaque = request.getOpaque();
    
        try { 
            // 创建一个该请求的ResponseFuture暂存在响应表中
            final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis, null, null);
            this.responseTable.put(opaque, responseFuture);
            final SocketAddress addr = channel.remoteAddress();
            // 发送请求
            channel.writeAndFlush(request).addListener(new ChannelFutureListener() { 
                // 请求发送成功的回调函数
                @Override
                public void operationComplete(ChannelFuture f) throws Exception { 
                    if (f.isSuccess()) { 
                        responseFuture.setSendRequestOK(true);
                        return;
                    } else { 
                        responseFuture.setSendRequestOK(false);
                    }
    
                    // 请求发送失败，移除缓存，响应设置为null
                    responseTable.remove(opaque);
                    responseFuture.setCause(f.cause());
                    responseFuture.putResponse(null);
                    log.warn("send a request command to channel <" + addr + "> failed.");
                }
            });
            // 同步等待响应结果，由响应线程唤醒
            RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
            if (null == responseCommand) { 
                // 响应为空但请求发送成功表示响应超时
                if (responseFuture.isSendRequestOK()) { 
                    throw new RemotingTimeoutException(RemotingHelper.parseSocketAddressAddr(addr), timeoutMillis,
                        responseFuture.getCause());
                } else { 
                    // 请求发送失败
                    throw new RemotingSendRequestException(RemotingHelper.parseSocketAddressAddr(addr), responseFuture.getCause());
                }
            }
    
            return responseCommand;
        } finally { 
            this.responseTable.remove(opaque);
        }
    }

`channel.writeAndFlush(request)`本质上也是异步请求，发送请求成功后就返回了，所以如果需要实现同步请求，就需要阻塞等待请求结果。  
实现过程是创建一个该请求的ResponseFuture暂存在响应表中，用于客户端发送数据后，服务端返回响应结果的时候，和客户端的请求正确匹配起来。具体过程是客户端发送请求时将<请求id, responseFuture>缓存起来，这时用CountDownLatch阻塞请求线程，等待响应结果。当netty客户端接收到服务端的响应后，响应线程根据响应数据中的请求id从缓存中取出responseFuture， 将响应结果设置到responseFuture中，然后通过CountDownLatch操作通知请求线程，将请求线程唤醒，请求线程通过responseFuture拿到响应结果进行处理。所以说到底还是异步请求，只是通过CountDownLatch将线程阻塞住等待结果。

这里分析了netty请求过程，后面涉及netty请求就不再分析了。

下面看看nameserver接收到心跳请求后如何处理。

### nameserver处理心跳包 ###

netty收到请求的处理入口在NettyRemotingServer中，刚好与发送请求的NettyRemotingClient对应。  
**NettyRemotingServer**

public class NettyRemotingServer extends NettyRemotingAbstract implements RemotingServer {     
        
        @ChannelHandler.Sharable
        class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> { 
    
            @Override
            protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception { 
                // netty服务端处理请求入口
                processMessageReceived(ctx, msg);
            }
        }
    }

调用父类方法  
**NettyRemotingAbstract**

public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception { 
        final RemotingCommand cmd = msg;
        if (cmd != null) { 
            switch (cmd.getType()) { 
                case REQUEST_COMMAND:
                    // 处理请求
                    processRequestCommand(ctx, cmd);
                    break;
                case RESPONSE_COMMAND:
                    // 处理响应
                    processResponseCommand(ctx, cmd);
                    break;
                default:
                    break;
            }
        }
    }

可以看到这有两种类型，处理请求可以理解，但为什么会有处理响应的？上面说到了，由于netty发送同步请求本质上还是异步请求，所以netty处理完请求后需要将响应结果重新发送给客户端。看源码发现NettyRemotingClient也继承了NettyRemotingAbstract

public class NettyRemotingClient extends NettyRemotingAbstract implements RemotingClient { 
    
        class NettyClientHandler extends SimpleChannelInboundHandler<RemotingCommand> { 
    
            @Override
            protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception { 
                processMessageReceived(ctx, msg);
            }
        }
    
    }

所以NettyRemotingClient这边的该方法是用来处理响应的。  
好了，接着回到server端处理接收到的请求

/** * 处理接收到的请求 * * @param ctx channel handler context. * @param cmd request command. */
    public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) { 
        // 获得属于该请求类型的Processor和Executor，没有则用默认的
        // 这里的注册请求没有专门的处理器和线程池，使用默认的
        final Pair<NettyRequestProcessor, ExecutorService> matched = this.processorTable.get(cmd.getCode());
        final Pair<NettyRequestProcessor, ExecutorService> pair = null == matched ? this.defaultRequestProcessor : matched;
        // 相当于请求id
        final int opaque = cmd.getOpaque();
    
        if (pair != null) { 
            Runnable run = new Runnable() { 
                @Override
                public void run() { 
                    try { 
                        doBeforeRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd);
                        // 对应处理器处理请求
                        final RemotingCommand response = pair.getObject1().processRequest(ctx, cmd);
                        doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd, response);
    
                        // 如果不是单向请求，需要将结果返回去，异步也会，只是不对响应进行处理
                        if (!cmd.isOnewayRPC()) { 
                            if (response != null) { 
                                // 将请求id设到响应中，用于客户端根据请求id找到对应的请求
                                response.setOpaque(opaque);
                                response.markResponseType();
                                try { 
                                    ctx.writeAndFlush(response);
                                } catch (Throwable e) { 
                                    log.error("process request over, but response failed", e);
                                    log.error(cmd.toString());
                                    log.error(response.toString());
                                }
                            } else { 
    
                            }
                        }
                    } catch (Throwable e) { 
                        log.error("process request exception", e);
                        log.error(cmd.toString());
    
                        if (!cmd.isOnewayRPC()) { 
                            final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_ERROR,
                                RemotingHelper.exceptionSimpleDesc(e));
                            response.setOpaque(opaque);
                            ctx.writeAndFlush(response);
                        }
                    }
                }
            };
    
            if (pair.getObject1().rejectRequest()) { 
                final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
                    "[REJECTREQUEST]system busy, start flow control for a while");
                response.setOpaque(opaque);
                ctx.writeAndFlush(response);
                return;
            }
    
            try { 
                // 外面再封装一层任务，任务中直接调用Runnable对象的run方法执行
                final RequestTask requestTask = new RequestTask(run, ctx.channel(), cmd);
                // 将任务提交到对应线程池中
                // 这里注册请求也是使用默认的线程池
                pair.getObject2().submit(requestTask);
            } catch (RejectedExecutionException e) { 
                if ((System.currentTimeMillis() % 10000) == 0) { 
                    log.warn(RemotingHelper.parseChannelRemoteAddr(ctx.channel())
                        + ", too many requests and system thread pool busy, RejectedExecutionException "
                        + pair.getObject2().toString()
                        + " request code: " + cmd.getCode());
                }
    
                if (!cmd.isOnewayRPC()) { 
                    final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
                        "[OVERLOAD]system busy, start flow control for a while");
                    response.setOpaque(opaque);
                    ctx.writeAndFlush(response);
                }
            }
        } else { 
            String error = " request type " + cmd.getCode() + " not supported";
            final RemotingCommand response =
                RemotingCommand.createResponseCommand(RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED, error);
            response.setOpaque(opaque);
            ctx.writeAndFlush(response);
            log.error(RemotingHelper.parseChannelRemoteAddr(ctx.channel()) + error);
        }
    }

接下来就需要使用对应的处理器创建一个线程提交到线程池中，由于注册请求没有专门的处理器和线程池，所以都是用默认的。  
**DefaultRequestProcessor**

/** * 处理netty接收到的请求 */
    @Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException { 
    
        if (ctx != null) { 
            log.debug("receive request, {} {} {}",
                request.getCode(),
                RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                request);
        }
    
        // 根据不同请求类型进行不同处理
        switch (request.getCode()) { 
            case RequestCode.PUT_KV_CONFIG:
                return this.putKVConfig(ctx, request);
            case RequestCode.GET_KV_CONFIG:
                return this.getKVConfig(ctx, request);
            case RequestCode.DELETE_KV_CONFIG:
                return this.deleteKVConfig(ctx, request);
            case RequestCode.QUERY_DATA_VERSION:
                return queryBrokerTopicConfig(ctx, request);
            // 注册broker和broke发送心跳的请求
            case RequestCode.REGISTER_BROKER:
                Version brokerVersion = MQVersion.value2Version(request.getVersion());
                if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) { 
                    return this.registerBrokerWithFilterServer(ctx, request);
                } else { 
                    // 注册broker
                    return this.registerBroker(ctx, request);
                }
            // broker正常下线，移除broker的注册信息
            case RequestCode.UNREGISTER_BROKER:
                return this.unregisterBroker(ctx, request);
            // 根据topic获取路由信息
            case RequestCode.GET_ROUTEINTO_BY_TOPIC:
                return this.getRouteInfoByTopic(ctx, request);
            case RequestCode.GET_BROKER_CLUSTER_INFO:
                return this.getBrokerClusterInfo(ctx, request);
            case RequestCode.WIPE_WRITE_PERM_OF_BROKER:
                return this.wipeWritePermOfBroker(ctx, request);
            case RequestCode.GET_ALL_TOPIC_LIST_FROM_NAMESERVER:
                return getAllTopicListFromNameserver(ctx, request);
            case RequestCode.DELETE_TOPIC_IN_NAMESRV:
                return deleteTopicInNamesrv(ctx, request);
            case RequestCode.GET_KVLIST_BY_NAMESPACE:
                return this.getKVListByNamespace(ctx, request);
            case RequestCode.GET_TOPICS_BY_CLUSTER:
                return this.getTopicsByCluster(ctx, request);
            case RequestCode.GET_SYSTEM_TOPIC_LIST_FROM_NS:
                return this.getSystemTopicListFromNs(ctx, request);
            case RequestCode.GET_UNIT_TOPIC_LIST:
                return this.getUnitTopicList(ctx, request);
            case RequestCode.GET_HAS_UNIT_SUB_TOPIC_LIST:
                return this.getHasUnitSubTopicList(ctx, request);
            case RequestCode.GET_HAS_UNIT_SUB_UNUNIT_TOPIC_LIST:
                return this.getHasUnitSubUnUnitTopicList(ctx, request);
            case RequestCode.UPDATE_NAMESRV_CONFIG:
                return this.updateConfig(ctx, request);
            case RequestCode.GET_NAMESRV_CONFIG:
                return this.getConfig(ctx, request);
            default:
                break;
        }
        return null;
    }

可以看到该默认的请求处理器也可以处理多种类型的请求，我们就看处理注册broker和broke发送心跳的请求

/** * 处理注册的请求 */
    public RemotingCommand registerBroker(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException { 
        // 构建响应的RemotingCommand
        final RemotingCommand response = RemotingCommand.createResponseCommand(RegisterBrokerResponseHeader.class);
        final RegisterBrokerResponseHeader responseHeader = (RegisterBrokerResponseHeader) response.readCustomHeader();
        final RegisterBrokerRequestHeader requestHeader =
            (RegisterBrokerRequestHeader) request.decodeCommandCustomHeader(RegisterBrokerRequestHeader.class);
    
        if (!checksum(ctx, request, requestHeader)) { 
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("crc32 not match");
            return response;
        }
    
        // topic路由配置信息
        TopicConfigSerializeWrapper topicConfigWrapper;
        if (request.getBody() != null) { 
            topicConfigWrapper = TopicConfigSerializeWrapper.decode(request.getBody(), TopicConfigSerializeWrapper.class);
        } else { 
            topicConfigWrapper = new TopicConfigSerializeWrapper();
            topicConfigWrapper.getDataVersion().setCounter(new AtomicLong(0));
            topicConfigWrapper.getDataVersion().setTimestamp(0);
        }
        // 向路由信息管理器实际注册broker信息
        RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(
            requestHeader.getClusterName(),
            requestHeader.getBrokerAddr(),
            requestHeader.getBrokerName(),
            requestHeader.getBrokerId(),
            requestHeader.getHaServerAddr(),
            topicConfigWrapper,
            null,
            ctx.channel()
        );
    
        responseHeader.setHaServerAddr(result.getHaServerAddr());
        responseHeader.setMasterAddr(result.getMasterAddr());
    
        byte[] jsonValue = this.namesrvController.getKvConfigManager().getKVListByNamespace(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG);
        response.setBody(jsonValue);
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

从NamesrvController中获取RouteInfoManager路由信息管理器来真正注册broker  
在真正注册之前，先来看看RouteInfoManager中即将用到的属性吧

private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
    
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    /** * 每个topic对应的队列 */
    private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
    /** * 维护每个主从架构中的broker信息 */
    private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
    /** * 维护每个broker集群中的每个broker名称，一般只会有一个集群 */
    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;

现在看真正执行注册的逻辑

/** * 向该路由信息管理器注册broker信息 */
    public RegisterBrokerResult registerBroker(
        final String clusterName,
        final String brokerAddr,
        final String brokerName,
        final long brokerId,
        final String haServerAddr,
        final TopicConfigSerializeWrapper topicConfigWrapper,
        final List<String> filterServerList,
        final Channel channel) { 
        RegisterBrokerResult result = new RegisterBrokerResult();
        try { 
            try { 
                // 获得写锁，防止并发修改路由表
                this.lock.writeLock().lockInterruptibly();
    
                // 取出该集群名下的broker名称集合，判断该broker所属集群是否存在，不存在则先创建一个
                Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
                if (null == brokerNames) { 
                    brokerNames = new HashSet<String>();
                    this.clusterAddrTable.put(clusterName, brokerNames);
                }
                // 将broker名称添加到clusterName下
                brokerNames.add(brokerName);
    
                // broker是否是第一次注册
                boolean registerFirst = false;
    
                BrokerData brokerData = this.brokerAddrTable.get(brokerName);
                if (null == brokerData) { 
                    // 没有broker主从架构信息，说明broker是第一次注册
                    registerFirst = true;
                    // 构建broker数据
                    brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());
                    this.brokerAddrTable.put(brokerName, brokerData);
                }
                Map<Long, String> brokerAddrsMap = brokerData.getBrokerAddrs();
                //Switch slave to master: first remove <1, IP:PORT> in namesrv, then add <0, IP:PORT>
                //The same IP:PORT must only have one record in brokerAddrTable
                Iterator<Entry<Long, String>> it = brokerAddrsMap.entrySet().iterator();
                while (it.hasNext()) { 
                    Entry<Long, String> item = it.next();
                    if (null != brokerAddr && brokerAddr.equals(item.getValue()) && brokerId != item.getKey()) { 
                        it.remove();
                    }
                }
    
                // 设置对应brokerId的地址，若存在，则更新，并且registerFirst设为true
                String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);
                registerFirst = registerFirst || (null == oldAddr);
    
                // broker是master
                if (null != topicConfigWrapper
                    && MixAll.MASTER_ID == brokerId) { 
                    // broker的topic路由信息改变了或首次注册，则需要创建或更新topic路由元数据，填充topicQueueTable
    				// 其实就是为默认主题自动注册路由信息，其中包含MixAll.DEFAULT_TOPIC的路由信息。当消息生产者发送
    				// 主题时，如果该主题为创建并且BrokerConfig的autoCreateTopicEnable为true时，将返回
    				// MixAll.DEFAULT_TOPIC的路由信息
                    if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())
                        || registerFirst) { 
                        ConcurrentMap<String, TopicConfig> tcTable =
                            topicConfigWrapper.getTopicConfigTable();
                        if (tcTable != null) { 
                            for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) { 
                                // 要创建或更新topic路由元数据，填充topicQueueTable
                                this.createAndUpdateQueueData(brokerName, entry.getValue());
                            }
                        }
                    }
                }
    
                // 心跳逻辑
                // 默认每隔30s收到心跳请求，生成一个新的BrokerLiveInfo，覆盖掉原来的BrokerLiveInfo，
                // BrokerLiveInfo里有一个lastUpdateTimestamp属性，表示最近收到心跳的时间
                BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddr,
                    new BrokerLiveInfo(
                        System.currentTimeMillis(),
                        topicConfigWrapper.getDataVersion(),
                        channel,
                        haServerAddr));
                if (null == prevBrokerLiveInfo) { 
                    log.info("new broker registered, {} HAServer: {}", brokerAddr, haServerAddr);
                }
    
                if (filterServerList != null) { 
                    if (filterServerList.isEmpty()) { 
                        this.filterServerTable.remove(brokerAddr);
                    } else { 
                        this.filterServerTable.put(brokerAddr, filterServerList);
                    }
                }
    
                // 如果是从节点，并且master信息已经注册过了，则将master地址设置到RegisterBrokerResult
                if (MixAll.MASTER_ID != brokerId) { 
                    String masterAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
                    if (masterAddr != null) { 
                        BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.get(masterAddr);
                        if (brokerLiveInfo != null) { 
                            result.setHaServerAddr(brokerLiveInfo.getHaServerAddr());
                            result.setMasterAddr(masterAddr);
                        }
                    }
                }
            } finally { 
                this.lock.writeLock().unlock();
            }
        } catch (Exception e) { 
            log.error("registerBroker Exception", e);
        }
    
        return result;
    }

具体的流程都写在注释里了

这里使用了读写锁来控制对路由表的并发读写，多个生产者可以同时来读取路由表，但同时只能一个broker来更改路由表。

### broker处理响应 ###

nameserver处理完注册请求后，会将结果响应回去，broker这边由NettyRemotingClient来处理响应，唤醒阻塞的请求线程，这时就会用到上面讲到的client来处理响应。

此时参数中的RemotingCommand就不是请求数据了，而是响应数据。

/** * 处理响应 */
    public void processResponseCommand(ChannelHandlerContext ctx, RemotingCommand cmd) { 
        final int opaque = cmd.getOpaque();
        // 根据id从正在等待响应的缓存中获取请求
        final ResponseFuture responseFuture = responseTable.get(opaque);
        if (responseFuture != null) { 
            // 设置响应数据
            responseFuture.setResponseCommand(cmd);
    
            responseTable.remove(opaque);
    
            // 异步会有InvokeCallback
            if (responseFuture.getInvokeCallback() != null) { 
                executeInvokeCallback(responseFuture);
            } else { 
                // 同步执行这里
                // 会使用countDownLatch.countDown()唤醒
                responseFuture.putResponse(cmd);
                responseFuture.release();
            }
        } else { 
            log.warn("receive response, but not matched any request, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
            log.warn(cmd.toString());
        }
    }

重新贴一下发送同步请求的代码：

public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request,
        final long timeoutMillis)
        throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException { 
        // 可以看作request id
        final int opaque = request.getOpaque();
    
        try { 
            final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis, null, null);
            this.responseTable.put(opaque, responseFuture);
            final SocketAddress addr = channel.remoteAddress();
            // 发送请求
            channel.writeAndFlush(request).addListener(new ChannelFutureListener() { 
                // 请求发送成功的回调函数
                @Override
                public void operationComplete(ChannelFuture f) throws Exception { 
                    if (f.isSuccess()) { 
                        responseFuture.setSendRequestOK(true);
                        return;
                    } else { 
                        responseFuture.setSendRequestOK(false);
                    }
    
                    // 请求发送失败，移除缓存，响应设置为null
                    responseTable.remove(opaque);
                    responseFuture.setCause(f.cause());
                    responseFuture.putResponse(null);
                    log.warn("send a request command to channel <" + addr + "> failed.");
                }
            });
            // 使用countDownLatch.await同步等待响应结果，由响应线程唤醒
            RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
            if (null == responseCommand) { 
                // 响应为空但请求发送成功表示响应超时
                if (responseFuture.isSendRequestOK()) { 
                    throw new RemotingTimeoutException(RemotingHelper.parseSocketAddressAddr(addr), timeoutMillis,
                        responseFuture.getCause());
                } else { 
                    // 请求发送失败
                    throw new RemotingSendRequestException(RemotingHelper.parseSocketAddressAddr(addr), responseFuture.getCause());
                }
            }
    
            return responseCommand;
        } finally { 
            this.responseTable.remove(opaque);
        }
    }

此时请求线程就会在31行代码处唤醒，去处理响应结果。  
![在这里插入图片描述][8f64431b5448ac75b5215f9356ddc1a2.png_pic_center]

## 参考资料 ##

《儒猿技术窝——从 0 开始带你成为消息中间件实战高手》

[918d062dccd22a173a588550cd1184f7.png_pic_center]: /images/20220828/8a3be2fc4d074a76b01873c3903edf9b.png
[8f64431b5448ac75b5215f9356ddc1a2.png_pic_center]: /images/20220828/044bfd6e78c24584a6082efa94361389.png