RocketMQ：深度剖析Broker启动流程原理、源码

￡神魔★判官ぃ 2023-10-02 12:36 62阅读 0赞

## 一、启动源 ##

我们先看一个broker的启动脚本：

nohup sh bin/mqbroker -n 100.10.100.100:9876 -c conf/broker.conf &

再看mqbroker启动脚本中：

sh ${ROCKETMQ_HOME}/bin/runbroker.sh org.apache.rocketmq.broker.BrokerStartup $@

至此，我们可以确定最终运行的是 BrokerStartup的main 方法，并将脚本参数以args的方式传入。

### apache 命令脚手架： ###

在我们启动broker时，即运行BrokerStartup\#main()方法时，会附带一些参数信息。使用命令脚手架可以很方便的让我们知道对应的参数信息。我们可以通过 -c `配置文件路径` ，来初始化配置。

## 二、启动流程 ##

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16]

### 1、BrokerStartup\#main()方法入口 ###

从方法的命令上来看，main方法主要做两件事，第一件事：创建BrokerController，第二件事：启动BrokerController。

public static void main(String[] args) {
        
        start(createBrokerController(args));
    }

#### 1.1、createBrokerController()方法。 ####

我们接着看BrokerStartup\#createBrokerController()方法都做了什么？  
**（1）设置MQ版本号、Netty客户端接送和接收缓存区大小；**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 1]  
**（2）注册命令脚手架；**

会获取到获取命令脚手架绑定过的参数。

public static CommandLine commandLine = null;

![在这里插入图片描述][3960303c4aee496ca72849ac449602ea.png]

##### 1）可以得知broker 启动时追加的参数信息 #####

>  *  \-n : 指定broker 的 namesrvAddr 地址；
>  *  \-h ：打印命令；
>  *  \-c ：指定配置文件的路径；
>  *  \-p ：启动时候日志打印配置信息；
>  *  \-m ：启动时候日志打印导入的配置信息。

**（3）绑定nettyServer的监听端口，即对外提供消息读写服务的端口；**  
![在这里插入图片描述][9d5f125607674288a23727695572a50f.png]  
**（4）创建四个配置类；并且读取broker.conf配置文件信息，利用反射将properties注入到配置类中；**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 2]

##### 1）Master和Slave启动时的区别？ #####

> 1.  accessMessageInMemoryMaxRatio消息占用内存的比率，超过这个值会进行内存置换。  
>     Master默认40%；Slave默认30%。
> 2.  brokerId机器的角色ID，master：0；Slave：大于0的数。

##### 2）四个配置类的作用？ #####

> 1.  BrokerConfig，封装绝大多数Broker的基本配置信息;
> 2.  NettyServerConfig，封装了broker作为对外提供消息读写操作的MQ服务器信息;
> 3.  NettyClientConfig，封装了broker作为NameServer的客户端的信息;
> 4.  MessageStoreConfig，封装Store的信息。

**下面这一堆也是给四个配置类（BrokerConfig、NettyServerConfig、NettyClientConfig、MessageStoreConfig）设置配置信息的；**

// 处理其他命令，注入到BrokerConfig中，比如 -n 命令
    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);
    }
    
    // 检查nameServer
    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);
        }
    }
    
    // 检查Broker角色
    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;
    }
    
    if (messageStoreConfig.isEnableDLegerCommitLog()) {
        
        brokerConfig.setBrokerId(-1);
    }
    
    // 高可用端口，10912
    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);

**（5）根据上面四个配置类，创建BrokerController；**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_16_color_FFFFFF_t_70_g_se_x_16]

##### 1）BrokerController的构造函数： #####

BrokerController是Broker启动逻辑的核心类，其构造函数中`包含很多内部服务类的初始化`。

public BrokerController(
        final BrokerConfig brokerConfig,
        final NettyServerConfig nettyServerConfig,
        final NettyClientConfig nettyClientConfig,
        final MessageStoreConfig messageStoreConfig
    ) {
        
    // BrokerStartup中准备的配置信息
    this.brokerConfig = brokerConfig;
    this.nettyServerConfig = nettyServerConfig;
    this.nettyClientConfig = nettyClientConfig;
    this.messageStoreConfig = messageStoreConfig;
    
    // Consumer消费进度记录管理类，会读取store/config/consumerOffset.json配置文件，其维护了一个offsetTable -- Map 结构
    this.consumerOffsetManager = new ConsumerOffsetManager(this);
    
    // 消息Topic维度的管理查询类， 管理Topic和Topic相关的配置关系，  会读取store/config/topics.json
    this.topicConfigManager = new TopicConfigManager(this);
    
    // Consumer端使用pull的方式向Broker拉取消息请求的处理类，  关联的业务code 为RequestCode.PULL_MESSAGE
    this.pullMessageProcessor = new PullMessageProcessor(this);
    
    // Consumer使用Push方式的长轮询机制拉取请求时，保存使用，当有消息到达时进行推送处理的类
    this.pullRequestHoldService = new PullRequestHoldService(this);
    
    // 有消息到达Broker时的监听器，回调pullRequestHoldService中的notifyMessageArriving()方法
    this.messageArrivingListener = new NotifyMessageArrivingListener(this.pullRequestHoldService);
    
    // 消费者ID变化监听器
    this.consumerIdsChangeListener = new DefaultConsumerIdsChangeListener(this);
    
    // 消费者管理类按照Group进行分组，对消费者的ID变化进行监听
    this.consumerManager = new ConsumerManager(this.consumerIdsChangeListener);
    
    // 消费者的过滤器管理类，按照Topic进行分类，  会读取store/config/consumerFilter.json
    this.consumerFilterManager = new ConsumerFilterManager(this);
    
    // 生产者管理类， 按照Topic进行分类
    this.producerManager = new ProducerManager();
    
    // 心跳连接处理类， 用于清除不活动的链接。
    this.clientHousekeepingService = new ClientHousekeepingService(this);
    
    // Console控制台获取Broker信息使用
    this.broker2Client = new Broker2Client(this);
    
    // 订阅关系管理类
    this.subscriptionGroupManager = new SubscriptionGroupManager(this);
    
    // Broker对外访问的API
    this.brokerOuterAPI = new BrokerOuterAPI(nettyClientConfig);
    
    // FilterServer管理类
    this.filterServerManager = new FilterServerManager(this);
    
    // Broker主从同步进度管理类
    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
    );
    }

这里实例化了很多个成员变量，我们挑几个重点介绍一下。

<table> 
 <thead> 
  <tr> 
   <th>字段类</th> 
   <th>作用</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>consumerOffsetManager</td> 
   <td>管理Consumer消费进度；会读取store/config/consumerOffset.json文件，其内部维护了一个Map结构offsetTable</td> 
  </tr> 
  <tr> 
   <td>topicConfigManager</td> 
   <td>管理Topic和Topic相关的配置关系，会读取store/config/topics.json文件</td> 
  </tr> 
  <tr> 
   <td>pullMessageProcessor</td> 
   <td>Consumer拉取消息时的请求包装类</td> 
  </tr> 
  <tr> 
   <td>pullRequestHoldService</td> 
   <td>Consumer采用Push方式，长轮询拉取消息服务</td> 
  </tr> 
  <tr> 
   <td>messageArrivingListener</td> 
   <td>采用长轮询机制拉取消息，当有消息到达Broker时的监听器；会调用到pullRequestHoldService中的notifyMessageArriving方法</td> 
  </tr> 
  <tr> 
   <td>defaultConsumerIdsChangeListener</td> 
   <td>消费者id变化监听器,主要监听Consumer的注册和下线的事件</td> 
  </tr> 
  <tr> 
   <td>consumerFilterManager</td> 
   <td>消费者的过滤器管理类，按照Topic进行分类， 会读取store/config/consumerFilter.json</td> 
  </tr> 
  <tr> 
   <td>clientHousekeepingService</td> 
   <td>心跳连接处理服务， 用于清除不活动的链接</td> 
  </tr> 
  <tr> 
   <td>subscriptionGroupManager</td> 
   <td>订阅关系管理类</td> 
  </tr> 
  <tr> 
   <td>slaveSynchronize</td> 
   <td>Broker主从同步进度管理类</td> 
  </tr> 
 </tbody> 
</table>

**（6）初始化BrokerController；**  
![在这里插入图片描述][0ac1ba70690546fbbb692fdcea9a75f8.png]  
在`BrokerStartup`的`createBrokerController`方法中，创建完BrokerController对象之后会马上调用其initialize方法对BrokerController进行初始化。这里我们梳理一下它主要做的事情：

\**1、加载服务器config/目录下的所有配置文件、日志文件。比如：Topic相关配置、Consumer消费消息进度情况、Consumer订阅关系、Consumer过滤关系，CommitLog、ConsumeQueue日志文件。*  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 3]

##### 1）注意这里的实例化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;
        // 初始化分配MappedFile文件服务
        this.allocateMappedFileService = new AllocateMappedFileService(this);
        // 初始化CommitLog存储服务
        if (messageStoreConfig.isEnableDLegerCommitLog()) {
        
            this.commitLog = new DLedgerCommitLog(this);
        } else {
        
            this.commitLog = new CommitLog(this);
        }
        // 初始化消费队列信息
        this.consumeQueueTable = new ConcurrentHashMap<>(32);
    
        // 初始化 队列文件ConsumeQueue刷盘服务
        this.flushConsumeQueueService = new FlushConsumeQueueService();
        // 初始化  清理CommitLog文件服务
        this.cleanCommitLogService = new CleanCommitLogService();
        // 初始化  清理ConsumeQueue文件服务
        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;
        }
        // 初始化  根据CommitLog文件构建ConsumeQueue、IndexFile文件服务
        this.reputMessageService = new ReputMessageService();
    
        this.scheduleMessageService = new ScheduleMessageService(this);
    
        this.transientStorePool = new TransientStorePool(messageStoreConfig);
    
        if (messageStoreConfig.isTransientStorePoolEnable()) {
        
            this.transientStorePool.init();
        }
    
        // 启动分配MappedFile文件线程
        this.allocateMappedFileService.start();
    
        // 启动队列索引文件服务， 空实现
        this.indexService.start();
    
        this.dispatcherList = new LinkedList<>();
        this.dispatcherList.addLast(new CommitLogDispatcherBuildConsumeQueue());
        this.dispatcherList.addLast(new CommitLogDispatcherBuildIndex());
    
        File file = new File(StorePathConfigHelper.getLockFile(messageStoreConfig.getStorePathRootDir()));
        MappedFile.ensureDirOK(file.getParent());
        lockFile = new RandomAccessFile(file, "rw");
    }

它又初始化了一堆服务，但是注意，它居然在构造函数中启动了两个服务：indexService（队列索引文件服务）、allocateMappedFileService（分配MappedFile文件服务）。  
`indexServer`.start()是一个空的方法，pass；我们继续看`allocateMappedFileService`  
![在这里插入图片描述][1eb425b3754f4561b9ca16763db99460.png]  
只要服务处于非停止状态，就一直循环做`MMAP内存映射文件操作`；大家一起默念：`零拷贝`、`零拷贝`、`零拷贝`。

**2、启动Broker客户端NettyServer服务，创建一些线程池用于处理各种API请求，比如：消息生成、消息消费等；**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 4]  
**3、注册事件到Broker客户端上，然后将API事件和对应的线程池绑定到一个Processor中；**  
![在这里插入图片描述][368bfbc96af34a6e980059418e5cd8f7.png]  
比如将生产消息事件和生产消息线程池绑定到sendProcessor上。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 5]  
**4、启动一些定时任务，比如：每天记录Broker的状态、每5秒持久化offset到JSON文件中、每3分钟清理掉有问题的Consumer、每秒打印Send/Pull/Query队列信息等**

final long initialDelay = UtilAll.computeNextMorningTimeMillis() - System.currentTimeMillis();
        final long period = 1000 * 60 * 60 * 24;
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        
            @Override
            public void run() {
        
                try {
        
                    // 每天记录broker的状态日志
                    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 {
        
                    // 每5秒持久化offset到JSON文件中
                    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 {
        
                    // 每10s持久化类过滤器
                    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 {
        
                    // 每3分钟，清理掉有问题的consumer
                    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 {
        
                        // 每2分钟获取最新的NameServer信息
                        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);
            }
        }

**5、使用Java的SPI机制初始化三个服务，比如：事务相关的服务、消息权限相关的服务、RPC调用钩子相关服务；**  
![在这里插入图片描述][4a95f8f1f7bd467f8b632ea74e3d6b7a.png]  
initialTransaction()中会动态加载TransactionalMessageService和AbstractTransactionalMessageCheckListener的实现类。以初始化事务消息服务、关联BrokerController到事务消息的回查监听器上；此外，会初始化transactionalMessageCheckService事务消息回查服务，每隔一分钟执行一次事务消息回查check。

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());
        }
        // 给事务消息的回查监听器关联BrokerController
        this.transactionalMessageCheckListener.setBrokerController(this);
        // 初始化事务消息回查服务，该服务会周期性的检查事务消息
        this.transactionalMessageCheckService = new TransactionalMessageCheckService(this);
    }

![在这里插入图片描述][06ee8ed40a304b1d8fa57adf2656819c.png]  
**（7）注册BrokerController关闭的构造方法。**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 6]  
至此，BrokerController的初始化完成。我们接着看BrokerController的start()方法。

#### 1.2、start(createBrokerController(args));方法，即：BrokerController.start()方法。 ####

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 7]

public void start() throws Exception {
        
        if (this.messageStore != null) {
        
            // 启动消息存储服务DefaultMessageStore，其会对/store/lock文件加锁，
            // 以确保在broker运行期间只有一个broker实例操作/store目录
            this.messageStore.start();
        }
    
        if (this.remotingServer != null) {
        
            // 启动Netty服务监听10911端口，对外提供服务（消息生产、消费）
            this.remotingServer.start();
        }
    
        if (this.fastRemotingServer != null) {
        
            // 监听10909端口
            this.fastRemotingServer.start();
        }
    
        if (this.fileWatchService != null) {
        
            // fileWatchService与TLS有关，todo tls解析
            this.fileWatchService.start();
        }
    
        if (this.brokerOuterAPI != null) {
        
            // 启动Netty客户端netty，broker使用其向外发送数据，比如：向NameServer上报心跳、topic信息。
            this.brokerOuterAPI.start();
        }
    
        if (this.pullRequestHoldService != null) {
        
            // 长轮询机制hold住拉取消息请求的服务
            this.pullRequestHoldService.start();
        }
    
        if (this.clientHousekeepingService != null) {
        
            // 每10s检查一遍非活动的连接服务
            this.clientHousekeepingService.start();
        }
    
        if (this.filterServerManager != null) {
        
            this.filterServerManager.start();
        }
    
        if (!messageStoreConfig.isEnableDLegerCommitLog()) {
        
            // 处理HA
            startProcessorByHa(messageStoreConfig.getBrokerRole());
            // 启动定时任务，定时与slave机器同步数据，同步的内容包括配置，消费位移等
            handleSlaveSynchronize(messageStoreConfig.getBrokerRole());
            // 向所有的nameserver发送本机所有的主题数据；
            // 包括主题名、读队列个数、写队列个数、队列权限、是否有序等
            this.registerBrokerAll(true, false, true);
        }
    
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        
    
            @Override
            public void run() {
        
                try {
        
                    // 定时向NameServer注册Broker，最小每10s。
                    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) {
        
            // Broker信息统计，这个没有具体的实现；所以暂时不用管
            this.brokerStatsManager.start();
        }
    
        if (this.brokerFastFailure != null) {
        
            // Broker对请求队列中的请求进行快速失败，返回`Broker繁忙、请稍后重试`信息
            this.brokerFastFailure.start();
        }
    }

啊，这里面又是启动了一堆服务，比如：DefaultMessageStore消息存储服务、Netty客户端服务（对外提供消息生产、消费服务）、Netty服务端服务（向NameServer上报心跳、topic信息）、长轮询机制Hold住pullRequest请求服务、定时清理不活动连接的服务等。除了启动服务之外，又是我们熟悉的定时任务。启动定时任务至少每10s定时向NameServer注册当前Broker。

##### 1）startProcessorByHa(messageStoreConfig.getBrokerRole()); #####

我们看一下关于处理HA的逻辑：  
![在这里插入图片描述][69527fa6df5748389da678868342f319.png]  
当Broker部署Slave的时候、并且transactionalMessageCheckService不为空；会启动事务消息回查服务，也就是开一个线程、线程内while(true)循环跑，直到程序停止；其逻辑是默认每分钟执行一次，我们可以通过transactionCheckInterval参数进行设置。而`事务消息回查服务`的初始化正发生在createBrokerController的过程中。

所以我们在聊事务消息的时候，可以把这块事务消息回查的源码输出一下。

##### 2）registerBrokerAll() #####

关于RocketMQ中的发送心跳逻辑，我们在之后的文章中统一介绍。大家可以参考下述源码注释。  
**BrokerController\#registerBrokerAll()**

public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway, boolean forceRegister) {
        
        // 序列化当前Broker中的topic信息
        // 1.得到TopicConfigManager 中维护的topicConfigTable map结构信息，并进行序列化
        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(),
                // broker名称
                this.brokerConfig.getBrokerName(),
                // brokerID ,0表示master
                this.brokerConfig.getBrokerId(),
                this.brokerConfig.getRegisterBrokerTimeoutMills())) {
        
            // 向所有的NameServer上报信息
            doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);
        }
    }

**BrokerController\#doRegisterBrokerAll()**

private void doRegisterBrokerAll(boolean checkOrderConfig, boolean oneway,
                                     TopicConfigSerializeWrapper topicConfigWrapper) {
        
        // 调用BrokerOutAPI注册所有Broker
        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\#registerBrokerAll()**

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 = new CopyOnWriteArrayList<>();
        // 获取所有的NameServer地址
        List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
        if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
        
    
            // 封装心跳包Header， 上报到NameServer的类型为RequestCode.REGISTER_BROKER
            final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
            requestHeader.setBrokerAddr(brokerAddr);
            requestHeader.setBrokerId(brokerId);
            requestHeader.setBrokerName(brokerName);
            requestHeader.setClusterName(clusterName);
            requestHeader.setHaServerAddr(haServerAddr);
            requestHeader.setCompressed(compressed);
    
            // 封装心跳包Body
            RegisterBrokerBody requestBody = new RegisterBrokerBody();
            // Topic信息
            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());
            for (final String namesrvAddr : nameServerAddressList) {
        
                // 4个核心线程、10个最大线程，阻塞队列：ArrayBlockingQueue---容量32、有界
                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 {
        
                // 超时等待，避免卡死。
                countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
        
            }
        }
    
        return registerBrokerResultList;
    }

#### 问：Broker启动时开启了几个端口号？分别是什么作用？ ####

**Broker在启动的时候会开三个端口：**

1.  10911 接收消息推送的端口。
2.  10912 高可用的端口。
3.  10909 推送消息的VIP端口

## 三、总结 ##

1.  当我们运行mqbroker命令的时候，会进入到BrokerStartup的main()方法中，并将脚本参数传入，使用apache命令脚手架解析对应的参数信息，比如 -c 配置文件路径。
2.  从最外围来讲，启动Broker分为两步，第一步创建并初始化BrokerController、第二步启动BrokerController。
3.  每一步都会初始化一些其他的服务、启动一些定时任务。

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/b7a63e8997c44d10a0e4d8d84bb84226.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 1]: https://img-blog.csdnimg.cn/65ed75777e0b4860b61cfe1321589cee.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[3960303c4aee496ca72849ac449602ea.png]: https://img-blog.csdnimg.cn/3960303c4aee496ca72849ac449602ea.png
[9d5f125607674288a23727695572a50f.png]: https://img-blog.csdnimg.cn/9d5f125607674288a23727695572a50f.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 2]: https://img-blog.csdnimg.cn/572e17ad08ad40efb7304b81ea9b24f9.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_16_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/9d6d44deab0c43c3a876b11c1e837423.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_16,color_FFFFFF,t_70,g_se,x_16
[0ac1ba70690546fbbb692fdcea9a75f8.png]: https://img-blog.csdnimg.cn/0ac1ba70690546fbbb692fdcea9a75f8.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 3]: https://img-blog.csdnimg.cn/797859c71ff144b699a04710d0c40063.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[1eb425b3754f4561b9ca16763db99460.png]: https://img-blog.csdnimg.cn/1eb425b3754f4561b9ca16763db99460.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 4]: https://img-blog.csdnimg.cn/f41c5bf857a142b2a5d91b040d1fac37.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[368bfbc96af34a6e980059418e5cd8f7.png]: https://img-blog.csdnimg.cn/368bfbc96af34a6e980059418e5cd8f7.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 5]: https://img-blog.csdnimg.cn/c35b95cc61554fc8813dcc0b27360f14.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[4a95f8f1f7bd467f8b632ea74e3d6b7a.png]: https://img-blog.csdnimg.cn/4a95f8f1f7bd467f8b632ea74e3d6b7a.png
[06ee8ed40a304b1d8fa57adf2656819c.png]: https://img-blog.csdnimg.cn/06ee8ed40a304b1d8fa57adf2656819c.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 6]: https://img-blog.csdnimg.cn/31284d82c12d47e992a73a88c8999b28.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 7]: https://img-blog.csdnimg.cn/fe1798f7166f4dff83158af4301fcddd.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[69527fa6df5748389da678868342f319.png]: https://img-blog.csdnimg.cn/69527fa6df5748389da678868342f319.png