【RocketMQ Broker 相关源码】- broker 启动源码（1）

---

本文章基于 RocketMQ 4.9.3

1. 前言

• 【RocketMQ】- 源码系列目录

在前面《RocketMQ NameServer》文章中我们介绍了 NameServer 的启动流程，这篇文章开始就看下 Broker 的启动源码，作为 RocketMQ 的两大核心之一，看看 Broker 启动又做了什么。

2. 概述

Broker 主要负责消息的存储、投递和查询以及服务高可用保证，为了实现这些功能，Broker 包含了以下几个重要子模块。

1. Remoting Module： 整个 Broker 的实体，负责处理来自客户端的请求。
2. Client Manager： 负责管理客户端(Producer/Consumer)和维护 Consumer 的 Topic 订阅信息。
3. Store Service： 提供方便简单的 API 接口处理消息存储、索引构建、消息查询功能。
4. HA Service： 高可用服务，提供 Master Broker 和 Slave Broker 之间的数据同步功能，实现主从同步。
5. Index Service：根据特定的 Message key 对投递到 Broker 的消息进行索引构建，以提供消息的快速查询，可以根据 Keys 或者根据时间戳快速查询。

此外， Broker 还包括很多定时任务，比如消费者偏移量持久化、消费者组过滤配置持久化、broker 保护、NameServer 地址拉取服务 …

3. BrokerStartup.main

broker 启动入口在 BrokerStartup.main 方法中，而在启动前（windows）需要在环境变量中设置 ROCKET_HOME。

下面可以看到源码中启动逻辑就是先通过 createBrokerController 创建出 BrokerController，然后再通过 start 方法启动。

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

4. createBrokerController

/*** 创建 BrokerController* @param args* @return*/
public static BrokerController createBrokerController(String[] args) {// 1. 设置 RocketMQ 版本为 4.9.3System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));try {// 2. 解析启动命令行的参数, 比如 nohup sh bin/mqbroker -n 127.0.0.1:9876 -c conf/broker.conf > logs/broker.log 2>&1 &Options options = ServerUtil.buildCommandlineOptions(new Options());commandLine = ServerUtil.parseCmdLine("mqbroker", args, buildCommandlineOptions(options),new PosixParser());if (null == commandLine) {// 如果解析不出命令行, 就直接退出System.exit(-1);}// 创建 broker 配置、Netty 服务端配置、Netty 客户端配置final BrokerConfig brokerConfig = new BrokerConfig();final NettyServerConfig nettyServerConfig = new NettyServerConfig();final NettyClientConfig nettyClientConfig = new NettyClientConfig();// 3. 设置是否开启 TLSnettyClientConfig.setUseTLS(Boolean.parseBoolean(System.getProperty(TLS_ENABLE,String.valueOf(TlsSystemConfig.tlsMode == TlsMode.ENFORCING))));// 4. Netty 服务端监听端口是 10911nettyServerConfig.setListenPort(10911);// 5. 创建 DefaultMessageStore 存储配置final MessageStoreConfig messageStoreConfig = new MessageStoreConfig();// 如果当前 Broker 是从节点if (BrokerRole.SLAVE == messageStoreConfig.getBrokerRole()) {// 设置 broker 最大可用物理内存, 一般来说 RocketMQ 为了提高性能都会把最新 【物理内存 * accessMessageInMemoryMaxRatio(默认 40)】 大小的消息放到内存中,// 但是如果是从节点这个值就是 30, 设置的小一点int ratio = messageStoreConfig.getAccessMessageInMemoryMaxRatio() - 10;messageStoreConfig.setAccessMessageInMemoryMaxRatio(ratio);}// 6. 判断命令行是否设置了 -c, -c 后面的就是 broker.confif (commandLine.hasOption('c')) {// 获取配置文件String file = commandLine.getOptionValue('c');if (file != null) {// 从文件中获取配置信息到 properties 中configFile = file;InputStream in = new BufferedInputStream(new FileInputStream(file));properties = new Properties();properties.load(in);// 首先设置一部分配置到环境变量中properties2SystemEnv(properties);// 然后将剩余的配置设置到 brokerConfig、nettyServerConfig、nettyClientConfig、messageStoreConfig 中MixAll.properties2Object(properties, brokerConfig);MixAll.properties2Object(properties, nettyServerConfig);MixAll.properties2Object(properties, nettyClientConfig);MixAll.properties2Object(properties, messageStoreConfig);// 设置文件到 BrokerPathConfigHelper#brokerConfigPathBrokerPathConfigHelper.setBrokerConfigPath(file);in.close();}}// 将启动命令中设置的配置设置到 brokerConfig 中MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), brokerConfig);// 7. 如果环境变量没有设置 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);}// 8. 获取 NameServer 地址, 在 broker.conf 中通过 namesrvAddr=127.0.0.1:9876 设置String namesrvAddr = brokerConfig.getNamesrvAddr();if (null != namesrvAddr) {try {// 根据 ';' 分割, 可以设置多个 NameServer 的String[] addrArray = namesrvAddr.split(";");for (String addr : addrArray) {// 这里其实是校验 NameServer 格式是不是根据 : 分割, 也就是是不是 ip + ":" + port 的格式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);}}// 9. 然后根据当前 broker 的不同角色设置角色 IDswitch (messageStoreConfig.getBrokerRole()) {case ASYNC_MASTER:case SYNC_MASTER:// 主节点的 brokerID = 0brokerConfig.setBrokerId(MixAll.MASTER_ID);break;case SLAVE:// 从节点的 brokerID > 0if (brokerConfig.getBrokerId() <= 0) {System.out.printf("Slave's brokerId must be > 0");System.exit(-3);}break;default:break;}// 如果使用了 Dleger 高可用, 不需要设置 brokerID, 由 Dleger 自动主从同步if (messageStoreConfig.isEnableDLegerCommitLog()) {brokerConfig.setBrokerId(-1);}// 11. 设置 HA 高可用端口, 就是 10912, 用于主从同步的messageStoreConfig.setHaListenPort(nettyServerConfig.getListenPort() + 1);// 12. 设置日志信息LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();JoranConfigurator configurator = new JoranConfigurator();configurator.setContext(lc);lc.reset();System.setProperty("brokerLogDir", "");if (brokerConfig.isIsolateLogEnable()) {System.setProperty("brokerLogDir", brokerConfig.getBrokerName() + "_" + brokerConfig.getBrokerId());}if (brokerConfig.isIsolateLogEnable() && messageStoreConfig.isEnableDLegerCommitLog()) {System.setProperty("brokerLogDir", brokerConfig.getBrokerName() + "_" + messageStoreConfig.getdLegerSelfId());}configurator.doConfigure(brokerConfig.getRocketmqHome() + "/conf/logback_broker.xml");// 13. 如果配置了 -p, 就是命令行里面设置了 -p, 就打印配置的日志, 方便调试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')) {// -m 就是只打印 @ImportantField 注解的字段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);// 12. 创建 BrokerControllerfinal BrokerController controller = new BrokerController(brokerConfig,nettyServerConfig,nettyClientConfig,messageStoreConfig);// 将 broker.config 里面的配置读取到 BrokerController#Configuration#allConfigs 中controller.getConfiguration().registerConfig(properties);// 初始化 BrokerControllerboolean initResult = controller.initialize();if (!initResult) {// 失败了直接退出controller.shutdown();System.exit(-3);}// 设置 broker 关闭时候的回调钩子Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {private volatile boolean hasShutdown = false;private AtomicInteger shutdownTimes = new AtomicInteger(0);@Overridepublic void run() {synchronized (this) {log.info("Shutdown hook was invoked, {}", this.shutdownTimes.incrementAndGet());if (!this.hasShutdown) {this.hasShutdown = true;// 当前时间long beginTime = System.currentTimeMillis();// 关闭 broker 启动的一堆服务和关闭线程池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;
}

这里面的创建逻辑和 NameServer 的很类似，首先就是设置了 RocketMQ 版本为 4.9.3，然后开始解析命令行参数并根据命令行参数创建 broker 配置、Netty 服务端配置、Netty 客户端配置。值得注意的是 Broker 这里既可以作为服务端（10911）跟生产者、消费者通信，又可以作为客户端跟 NameServer 进行通信。

然后就是 broker 角色如果是从节点，那么设置 broker 最大可用物理内存，一般来说 RocketMQ 为了提高性能都会把最新 【物理内存 * accessMessageInMemoryMaxRatio(默认 40)】 大小的消息放到内存中，但是如果是从节点这个值就是 30, 设置的小一点。

broker 启动命令行例子如 nohup sh bin/mqbroker -n 127.0.0.1:9876 -c conf/broker.conf > logs/broker.log 2>&1 &，-c 后面跟的是配置文件 broker.conf 的路径，然后将这部分文件读到 properties 中，接着这将配置文件里面的配置分别设置到环境变量和不同的 config 中，如 brokerConfig、nettyServerConfig、nettyClientConfig、messageStoreConfig。

当然有一个很有意思的点，这里 NameServer 和 Broker 的设置配置都是通过反射去完成，就是通过找到对应属性的 set 方法来设置属性值，大家可以看下下面的代码。

public static void properties2Object(final Properties p, final Object object) {Method[] methods = object.getClass().getMethods();for (Method method : methods) {String mn = method.getName();if (mn.startsWith("set")) {try {String tmp = mn.substring(4);String first = mn.substring(3, 4);String key = first.toLowerCase() + tmp;String property = p.getProperty(key);if (property != null) {Class<?>[] pt = method.getParameterTypes();if (pt != null && pt.length > 0) {String cn = pt[0].getSimpleName();Object arg = null;if (cn.equals("int") || cn.equals("Integer")) {arg = Integer.parseInt(property);} else if (cn.equals("long") || cn.equals("Long")) {arg = Long.parseLong(property);} else if (cn.equals("double") || cn.equals("Double")) {arg = Double.parseDouble(property);} else if (cn.equals("boolean") || cn.equals("Boolean")) {arg = Boolean.parseBoolean(property);} else if (cn.equals("float") || cn.equals("Float")) {arg = Float.parseFloat(property);} else if (cn.equals("String")) {arg = property;} else {continue;}method.invoke(object, arg);}}} catch (Throwable ignored) {}}}
}

继续回到 createBrokerController 的代码，继续往下看，可以看到如果环境变量没有设置 ROCKETMQ_HOME，直接退出，所以如果是 windows 环境下要启动 RocketMQ 调试源码就需要在 IDEA 环境变量设置上 ROCKETMQ_HOME。

剩下的源码就直接看注释即可，因为比较核心的逻辑还是在下面的 start 方法。

5. BrokerController#initialize

在 createBrokerController 方法最后创建出 BrokerController 之后会调用 initialize 方法去初始化这个 controller，这里面要做的东西就非常多了，就不直接把全部代码直接放出来，而是一行一行去看。

// 1. 加载配置文件到内存中
//    1.1 加载 ${user.home}/store/config/topics.json(默认) topic 信息配置文件到 topicConfigTable 中
boolean result = this.topicConfigManager.load();
//    1.2 加载 ${user.home}/store/config/consumerOffset.json(默认) 消费者消费偏移量配置文件到 offsetTable 中
result = result && this.consumerOffsetManager.load();
//    1.3 加载 ${user.home}/store/config/subscriptionGroup.json.(默认) 消费者组订阅配置到 subscriptionGroupTable 中
result = result && this.subscriptionGroupManager.load();
//    1.4 加载 ${user.home}/store/config/consumerFilter.json.(默认) 消费者组过滤配置到 subscriptionGroupTable 中
result = result && this.consumerFilterManager.load();

首先第一部分，先加载配置文件里面的配置到内存中，这里的配置文件指的是 ${user.home}/store/config 文件夹，加载的配置文件就是下面四个：

• topics.json： topic 信息
• consumerOffset.json： 消费者消费偏移量
• subscriptionGroup.json： 消费者组订阅配置
• consumerFilter.json： 消费者组过滤配置，SQL92 过滤专用

// 2. 如果配置都加载成功了, 下面初始化 DefaultMessageStore
if (result) {try {// 创建 DefaultMessageStorethis.messageStore =new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener,this.brokerConfig);// 在 4.5 版本之后引入了基于 Raft 协议的 DLeger 存储库, 同时使用 DLeger commitlog 代替了原来的 CommitLog, 使得// CommitLog 拥有了选举复制能力, 然后通过角色透传的方式, raft 角色透传给外部 broker 角色, leader 对应原来的 master,// follower 和 candidate 对应原来的 slave, 如果开启了 Dleger, 那么主从复制这些逻辑都会交给 Dleger 完成, 默认是不开启if (messageStoreConfig.isEnableDLegerCommitLog()) {DLedgerRoleChangeHandler roleChangeHandler = new DLedgerRoleChangeHandler(this, (DefaultMessageStore) messageStore);((DLedgerCommitLog)((DefaultMessageStore) messageStore).getCommitLog()).getdLedgerServer().getdLedgerLeaderElector().addRoleChangeHandler(roleChangeHandler);}// broker 数据统计服务, 比如今天添加的消息数、昨天添加的消息数 ...this.brokerStats = new BrokerStats((DefaultMessageStore) this.messageStore);// load plugin// 加载消息存储插件, 默认是没有插件MessageStorePluginContext context = new MessageStorePluginContext(messageStoreConfig, brokerStatsManager, messageArrivingListener, brokerConfig);this.messageStore = MessageStoreFactory.build(context, this.messageStore);// 添加 SQL92 过滤的布隆过滤器分发类this.messageStore.getDispatcherList().addFirst(new CommitLogDispatcherCalcBitMap(this.brokerConfig, this.consumerFilterManager));} catch (IOException e) {result = false;log.error("Failed to initialize", e);}
}

接着第二步，当配置文件都加载完成了，就初始化 DefaultMessageStore，DefaultMessageStore 是 RocketMQ 存储的核心类了，大家也可以去看我之前写过的文章，就是 【RocketMQ 存储】 系列文章。而关于 Dleger 的文章，可以去看我的 RocketMQ 系列的 【RocketMQ 高可用】，不过一般来说都是不开启的，而且 RocketMQ 5.x 版本又对这块的逻辑做了重新处理，所以大家也可以去看官方文章：一文详解Apache RocketMQ 如何利用 Raft 进行高可用保障。

在前面的存储文章，特别是消息重放那一篇看源码就能看到了对消息分发实际上就是构建 ConsumeQueue、Index 索引以及构建 SQL 布隆过滤器索引，方便后期进行 SQL92 消息过滤，但是布隆过滤器这块还没有做文章来介绍，等后面到消费者那一块再去看吧，这里源码我也还没细看。

// 3. 加载 DefaultMessageStore 里面的配置文件, 如 CommitLog 文件、ConsumeQueue 文件、IndexFile 文件 ...
result = result && this.messageStore.load();

继续回到源码，加载 DefaultMessageStore 里面的配置文件, 如 CommitLog 文件、ConsumeQueue 文件、IndexFile 文件，这部分代码在下一篇文章给出来。

到这里可以看到上面返回的结果 result 字段，这个字段如果是 true，就表示前面的配置文件正常加载，下面处理 if 里面的逻辑，第一步就是创建 Netty 服务端配置。

// 4. 创建 NettyRemotingServer, broker 作为服务端监听端口 10911, 专门处理来自生产者和消费者的请求
this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.clientHousekeepingService);
// 克隆一份 Netty 服务端的配置文件
NettyServerConfig fastConfig = (NettyServerConfig) this.nettyServerConfig.clone();
// 修改下监听地址, 10909
fastConfig.setListenPort(nettyServerConfig.getListenPort() - 2);
// 5. 创建 fastRemotingServer, broker 作为服务端监听端口 10909, 专门处理内部的请求
this.fastRemotingServer = new NettyRemotingServer(fastConfig, this.clientHousekeepingService);

上面就是创建 remotingServer 和 fastRemotingServer，remotingServer 负责监听 10911 端口，fastRemotingServer 负责监听 10909 端口，接着往下看，下面会创建业务相关的线程池。

// 6. 创建发送消息的线程池
this.sendMessageExecutor = new BrokerFixedThreadPoolExecutor(this.brokerConfig.getSendMessageThreadPoolNums(),this.brokerConfig.getSendMessageThreadPoolNums(),1000 * 60,TimeUnit.MILLISECONDS,this.sendThreadPoolQueue,new ThreadFactoryImpl("SendMessageThread_"));// 6. 创建处理异步发送消息回调的线程池
this.putMessageFutureExecutor = new BrokerFixedThreadPoolExecutor(this.brokerConfig.getPutMessageFutureThreadPoolNums(),this.brokerConfig.getPutMessageFutureThreadPoolNums(),1000 * 60,TimeUnit.MILLISECONDS,this.putThreadPoolQueue,new ThreadFactoryImpl("PutMessageThread_"));// 7. 创建消息拉取线程池
this.pullMessageExecutor = new BrokerFixedThreadPoolExecutor(this.brokerConfig.getPullMessageThreadPoolNums(),this.brokerConfig.getPullMessageThreadPoolNums(),1000 * 60,TimeUnit.MILLISECONDS,this.pullThreadPoolQueue,new ThreadFactoryImpl("PullMessageThread_"));// 8. 创建消息回复线程池
this.replyMessageExecutor = new BrokerFixedThreadPoolExecutor(this.brokerConfig.getProcessReplyMessageThreadPoolNums(),this.brokerConfig.getProcessReplyMessageThreadPoolNums(),1000 * 60,TimeUnit.MILLISECONDS,this.replyThreadPoolQueue,new ThreadFactoryImpl("ProcessReplyMessageThread_"));// 9. 创建处理查询消息请求的线程池
this.queryMessageExecutor = new BrokerFixedThreadPoolExecutor(this.brokerConfig.getQueryMessageThreadPoolNums(),this.brokerConfig.getQueryMessageThreadPoolNums(),1000 * 60,TimeUnit.MILLISECONDS,this.queryThreadPoolQueue,new ThreadFactoryImpl("QueryMessageThread_"));// 10. 创建 AdminBrokerProcessor 处理器线程池
this.adminBrokerExecutor =Executors.newFixedThreadPool(this.brokerConfig.getAdminBrokerThreadPoolNums(), new ThreadFactoryImpl("AdminBrokerThread_"));// 11. 创建 ClientManageProcessor 处理器线程池, 处理一些客户端信息比如取消注册生产者、消费者的请求
this.clientManageExecutor = new ThreadPoolExecutor(this.brokerConfig.getClientManageThreadPoolNums(),this.brokerConfig.getClientManageThreadPoolNums(),1000 * 60,TimeUnit.MILLISECONDS,this.clientManagerThreadPoolQueue,new ThreadFactoryImpl("ClientManageThread_"));// 12. 创建用于心跳处理的线程池
this.heartbeatExecutor = new BrokerFixedThreadPoolExecutor(this.brokerConfig.getHeartbeatThreadPoolNums(),this.brokerConfig.getHeartbeatThreadPoolNums(),1000 * 60,TimeUnit.MILLISECONDS,this.heartbeatThreadPoolQueue,new ThreadFactoryImpl("HeartbeatThread_", true));// 13. 创建事务相关的线程池
this.endTransactionExecutor = new BrokerFixedThreadPoolExecutor(this.brokerConfig.getEndTransactionThreadPoolNums(),this.brokerConfig.getEndTransactionThreadPoolNums(),1000 * 60,TimeUnit.MILLISECONDS,this.endTransactionThreadPoolQueue,new ThreadFactoryImpl("EndTransactionThread_"));// 14. 创建消费者管理相关请求的线程池, 比如获取消费者消费偏移量请求
this.consumerManageExecutor =Executors.newFixedThreadPool(this.brokerConfig.getConsumerManageThreadPoolNums(), new ThreadFactoryImpl("ConsumerManageThread_"));

然后上面就是创建了几个线程池，这几个线程池就是处理消息查询、消息发送、消息拉取、消息回复、客户端取消注册请求、心跳处理 … 的线程池，那么这些线程池创建出来有什么用呢？看下面这行代码。

// 15. 注册处理器, RocketMQ 中发送请求的时候需要设置这个请求的 Code, 代表这个请求是个什么请求, 这里面就是注册这些 Code 的处理器,
//     比如生产者生产消息发送请求的 Code 是 SEND_MESSAGE, 这里面就会注册 SendMessageProcessor 来处理这个 Code, 同时也会设置
//     线程池去多线程处理这些请求
this.registerProcessor();

这里面就是注册消息处理器的逻辑，我们知道请求发送的时候都会有一个 code 来标注这是一个什么请求，比如当生产者发送消息的时候发送的请求的 code 就是 SEND_MESSAGE，那么在这个分发中就会注册一个 SendMessageProcessor 处理器去处理这个 code 请求，而处理生产者的发送请求不可能是单线程处理的，所以这种情况下就需要将 sendMessageExecutor 这个线程池也注册进去，简单来说就是使用线程池多线程处理，后面讲到生产者生产消息的时候这部分也会讲到。

那接着往下看，创建出线程池之后，开始启动一些定时任务，首先就是初始化 broker 统计服务, 初始化之后第二天 0 点才执行, 之后每隔 24 小时执行一次。这个任务就是每天记录一下前一天生产了和消费了多少消息到 BrokerStats 中，算是一个消息统计。

// 16. 初始化 broker 统计服务, 初始化之后第二天 0 点才执行, 之后每隔 24 小时执行一次
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {try {// 每天记录一下前一天生产了和消费了多少消息到 BrokerStats 中BrokerController.this.getBrokerStats().record();} catch (Throwable e) {log.error("schedule record error.", e);}}
}, initialDelay, period, TimeUnit.MILLISECONDS);

接着初始化消费者消费偏移量持久化服务, 初始化 10s 之后执行, 之后每 5s 执行一次，持久化消费者偏移量到 consumerOffset.json 和 consumerOffset.json.bak 文件，这个 .bak 文件是一个备份文件。

// 17. 初始化消费者消费偏移量持久化服务, 初始化 10s 之后执行, 之后每 5s 执行一次
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {try {// 持久化消费者偏移量到 consumerOffset.json 和 consumerOffset.json.bak 文件BrokerController.this.consumerOffsetManager.persist();} catch (Throwable e) {log.error("schedule persist consumerOffset error.", e);}}
}, 1000 * 10, this.brokerConfig.getFlushConsumerOffsetInterval(), TimeUnit.MILLISECONDS);

初始化消费者组过滤配置持久化服务, 初始化 10s 之后执行, 之后每 10s 执行一次，持久化消费者组过滤配置到 consumerFilter.json 和 consumerFilter.json.bak 文件。

// 18. 初始化消费者组过滤配置持久化服务, 初始化 10s 之后执行, 之后每 10s 执行一次
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {try {// 持久化消费者组过滤配置到 consumerFilter.json 和 consumerFilter.json.bak 文件BrokerController.this.consumerFilterManager.persist();} catch (Throwable e) {log.error("schedule persist consumer filter error.", e);}}
}, 1000 * 10, 1000 * 10, TimeUnit.MILLISECONDS);

初始化 broker 保护服务, 初始化 10s 之后执行, 之后每 10s 执行一次，保护 broker，在里面会校验如果这个消费者组的消息拉取进度落后最新消息超过 16G，说明有可能这个消费者组里面的消费者出问题了，不能正常消费消息，所以这时候就会将消费者组的订阅配置设置为禁止消费, 需要人为介入去修复。

// 19. 初始化 broker 保护服务, 初始化 10s 之后执行, 之后每 10s 执行一次
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {try {// 保护 broker, 在里面会校验如果这个消费者组的消息拉取进度落后最新消息超过 16G, 说明有可能这个消费者组里面的消// 费者出问题了, 不能正常消费消息, 所以这时候就会将消费者组的订阅配置设置为禁止消费, 需要人为介入去修复BrokerController.this.protectBroker();} catch (Throwable e) {log.error("protectBroker error.", e);}}
}, 3, 3, TimeUnit.MINUTES);

初始化线程池队列打印服务，初始化 10s 之后执行，之后每 1s 执行一次，打印消息发送、消息拉取、消息查询、事务结束处理线程池的队列大小。

// 20. 初始化线程池队列打印服务, 初始化 10s 之后执行, 之后每 1s 执行一次
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {try {// 打印消息发送、消息拉取、消息查询、事务结束处理线程池的队列大小BrokerController.this.printWaterMark();} catch (Throwable e) {log.error("printWaterMark error.", e);}}
}, 10, 1, TimeUnit.SECONDS);

初始化未重放的消息偏移量，初始化 10s 之后执行，之后每 1s 执行一次，打印消息重放服务还有多少 CommitLog 消息没有重放。

// 21. 初始化未重放的消息偏移量, 初始化 10s 之后执行, 之后每 1s 执行一次
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {try {// 打印消息重放服务还有多少 CommitLog 消息没有重放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);

接下来处理 NameServer，判断需不需要从地址服务器里面拉取 NameServer 地址，如果允许从地址服务器拉取地址并且 broker.conf 里面没有配置 NameServer 地址，就从地址服务器拉取 NameServer 地址，初始化 NameServer 地址拉取服务，初始化 10s 之后执行，之后每 120s 执行一次。

if (this.brokerConfig.getNamesrvAddr() != null) {
// 22. 更新 nameserver 地址到 namesrvAddrList
this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr());
log.info("Set user specified name server address: {}", this.brokerConfig.getNamesrvAddr());
} else if (this.brokerConfig.isFetchNamesrvAddrByAddressServer()) {
// 如果允许从地址服务器拉取地址并且 broker.conf 里面没有配置 NameServer 地址, 就从地址服务器拉取 NameServer 地址,
// 初始化 NameServer 地址拉取服务, 初始化 10s 之后执行, 之后每 120s 执行一次
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);
}

下面判断如果没有开启 Dleger 服务, 针对 broker 节点的不同（主节点、从节点）走不同逻辑，这一块逻辑里面的 updateMasterHAServerAddrPeriodically 参数我在 【RocketMQ 高可用】 这个系列文章的源码介绍里面已经介绍过了，大家可以去看下这几篇文章，就是有关 RocketMQ 的主从同步的。

// 23. 如果没有开启 Dleger 服务, 针对主从节点设置不同定时任务
if (!messageStoreConfig.isEnableDLegerCommitLog()) {// 如果当前 broker 是从节点if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {// 判断有没有配置 HA 高可用地址, 如果配置了就更新 HA 高可用地址, 同时设置 updateMasterHAServerAddrPeriodically 为 false,// updateMasterHAServerAddrPeriodically 用来配置是否在从节点 broker 启动的时候永久设置高可用地址, 也是从节点用来主从同步的，// 如果没用配置，就默认每一次注册 broker 信息到到 namesrv 都会更新当前从节点 broker 的高可用地址if (this.messageStoreConfig.getHaMasterAddress() != null && this.messageStoreConfig.getHaMasterAddress().length() >= 6) {this.messageStore.updateHaMasterAddress(this.messageStoreConfig.getHaMasterAddress());this.updateMasterHAServerAddrPeriodically = false;} else {// 这里就是没有配置this.updateMasterHAServerAddrPeriodically = true;}} else {// 如果当前 broker 是主节点, 初始化偏移量打印服务, 初始化 10s 之后执行, 之后每 60s 执行一次this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {try {// 打印主节点还有多少 CommitLog 消息没有同步到从节点(偏移量)BrokerController.this.printMasterAndSlaveDiff();} catch (Throwable e) {log.error("schedule printMasterAndSlaveDiff error.", e);}}}, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);}
}

接着注册监听器监听 TLS 相关证书文件的变化, 一旦证书文件发生变化, 就会重新加载服务器的 SSL 上下文, 以确保服务器使用的是最新的证书。

// 24. 注册监听器监听 TLS 相关证书文件的变化, 一旦证书文件发生变化, 就会重新加载服务器的 SSL 上下文, 以确保服务器使用的是最新的证书
if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {// Register a listener to reload SslContexttry {fileWatchService = new FileWatchService(new String[] {TlsSystemConfig.tlsServerCertPath,TlsSystemConfig.tlsServerKeyPath,TlsSystemConfig.tlsServerTrustCertPath},new FileWatchService.Listener() {boolean certChanged, keyChanged = false;@Overridepublic 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");}
}

最后初始化事务消息的相关服务，同时初始化 ACL 权限相关的服务和初始化 RPC 相关的钩子，这些东西是干什么用的，留到下一篇文章分析，这一批文章先展示下总体概述。

// 25. 初始化事务消息的相关服务
initialTransaction();
// 26. 初始化 ACL 权限相关的服务
initialAcl();
// 27. 初始化 RPC 钩子
initialRpcHooks();

6. BrokerStartup#start

public static BrokerController start(BrokerController controller) {try {// 启动 brokercontroller.start();String tip = "The broker[" + controller.getBrokerConfig().getBrokerName() + ", "+ controller.getBrokerAddr() + "] boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();if (null != controller.getBrokerConfig().getNamesrvAddr()) {tip += " and name server is " + controller.getBrokerConfig().getNamesrvAddr();}log.info(tip);System.out.printf("%s%n", tip);return controller;} catch (Throwable e) {e.printStackTrace();System.exit(-1);}return null;
}

这个方法就是调用 BrokerController.start 方法来启动，启动完成后，打印日志。

6.1 BrokerController#start

/*** 启动 broker, 实际就是启动一堆服务* @throws Exception*/
public void start() throws Exception {if (this.messageStore != null) {// 消息存储服务, DefaultMessageStore, 用于处理消息存储相关的逻辑this.messageStore.start();}// broker 一共启动了两个端口, 10911 和 10909, 其中 10909 也是所谓的 VIP 端口, 意思是不对外开放的, 专门用于处理一些内部的请求// 比如超时同步加锁请求、获取配置等等, VIP 通道跟普通通道没什么区别，但是 VIP 通道是平时很少用的，所以性能会很高if (this.remotingServer != null) {// 启动 Netty 服务端, 监听 10911 端口, Broker 作为服务端会接收来自生产者和消费者的请求this.remotingServer.start();}if (this.fastRemotingServer != null) {// 启动 fastRemotingServer 服务端, 监听 10909 端口, 也就是我们说的 VIP 端口this.fastRemotingServer.start();}// 启动文件监控器, 其实就是监听某一些文件 hash 值是否发生了更改, 如果变化了就调用 listener 的 onChanged 方法, 之前启动 NameServer// 的时候这玩意就是用来监听 TLS相关证书文件的变化, 一旦证书文件发生变化, 就会重新加载服务器的 SSL 上下文, 以确保服务器使用的是最新的证书if (this.fileWatchService != null) {this.fileWatchService.start();}// 启动 brokerOuterAPI, 就是 broker 作为客户端向 NameServer 发送的一些信息比如注册 broker 信息、topic 信息 ...// 上面启动了 RemotingServer, 这里会启动 RemotingClient, 所以 broker 既可以作为服务端也可以作为客户端if (this.brokerOuterAPI != null) {this.brokerOuterAPI.start();}// 消费者拉取请求挂起服务if (this.pullRequestHoldService != null) {this.pullRequestHoldService.start();}// 客户端心跳检测服务, 就是去检测不活跃的生产者、消费者、过滤器服务if (this.clientHousekeepingService != null) {this.clientHousekeepingService.start();}// 消息过滤服务, 以前的版本 broker 启动的时候都会在这里顺带启动一个消息过滤服务, 消费者可以将过滤的信息通过代码提交到这个过滤服务// 来进行消息的拉取过滤, 但是现在我记得应该是没有用了, 换成 SQL92 了, 找不到处理的 Codeif (this.filterServerManager != null) {this.filterServerManager.start();}// 这里就是启动一些事务相关的服务if (!messageStoreConfig.isEnableDLegerCommitLog()) {// 启动事务消息回查服务 transactionalMessageCheckServicestartProcessorByHa(messageStoreConfig.getBrokerRole());// 启动从节点数据同步服务handleSlaveSynchronize(messageStoreConfig.getBrokerRole());// 将 broker 信息注册到所有的 NameServerthis.registerBrokerAll(true, false, true);}// 启动定时服务, 初始化之后 10s 开始执行第一次, 之后每隔 30s 执行一次this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {try {// 将 broker 信息注册到所有的 NameServerBrokerController.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);// 启动 broker 信息统计服务, 比如统计 topic 数量、队列数量、消息总数 ...if (this.brokerStatsManager != null) {this.brokerStatsManager.start();}// 启动 broker 快速失败服务, 这里当 DefaultMessageStore 操作系统页繁忙的时候会将发送消息的请求全部返回一个 SYSTEM_BUSY 错误码// 也就是说没办法继续往 CommitLog 里面添加生产者生产的消息if (this.brokerFastFailure != null) {this.brokerFastFailure.start();}
}

BrokerController#start 方法里面就是启动了一堆服务，就是如下这些：

• 消息存储服务，DefaultMessageStore，用于处理消息存储相关的逻辑
• Netty 服务端, 监听 10911 端口, Broker 作为服务端会接收来自生产者和消费者的请求
• fastRemotingServer 服务端, 监听 10909 端口, 也就是我们说的 VIP 端口
• 文件监控器, 其实就是监听某一些文件 hash 值是否发生了更改, 如果变化了就调用 listener 的 onChanged 方法
• brokerOuterAPI, 就是 broker 作为客户端向 NameServer 发送的一些信息比如 broker 心跳信息、topic 信息 …
• 消费者拉取请求挂起服务，在后面消费者文章会讲到
• 客户端心跳检测服务, 就是去检测不活跃的生产者、消费者、过滤器服务
• 消息过滤服务, 以前的版本 broker 启动的时候都会在这里顺带启动一个消息过滤服务, 消费者可以将过滤的信息通过代码提交到这个过滤服务来进行消息的拉取过滤, 但是现在我记得应该是没有用了, 换成 SQL92 了, 找不到处理的 Code
• 启动事务消息回查服务 transactionalMessageCheckService
• 启动从节点数据同步服务
• 启动定时服务, 初始化之后 10s 开始执行第一次, 之后每隔 30s 执行一次，将 broker 信息注册到所有的 NameServer
• 启动 broker 信息统计服务, 比如统计 topic 数量、队列数量、消息总数 …
• 启动 broker 快速失败服务, 这里当 DefaultMessageStore 操作系统页繁忙的时候会将发送消息的请求全部返回一个 SYSTEM_BUSY 错误码，也就是说没办法继续往 CommitLog 里面添加生产者生产的消息

7. BrokerController#shutdown

还记得 createBrokerController 方法最后添加的钩子方法吗，当 broker 关闭的时候会在钩子方法中回调 shutdown 方法，这个方法就是将前面启动的一堆服务给关闭了，下面我给出源码，注释就不一一写了，大家可以看上面这几个小结。

public void shutdown() {if (this.brokerStatsManager != null) {this.brokerStatsManager.shutdown();}if (this.clientHousekeepingService != null) {this.clientHousekeepingService.shutdown();}if (this.pullRequestHoldService != null) {this.pullRequestHoldService.shutdown();}if (this.remotingServer != null) {this.remotingServer.shutdown();}if (this.fastRemotingServer != null) {this.fastRemotingServer.shutdown();}if (this.fileWatchService != null) {this.fileWatchService.shutdown();}if (this.messageStore != null) {this.messageStore.shutdown();}this.scheduledExecutorService.shutdown();try {this.scheduledExecutorService.awaitTermination(5000, TimeUnit.MILLISECONDS);} catch (InterruptedException e) {}this.unregisterBrokerAll();if (this.sendMessageExecutor != null) {this.sendMessageExecutor.shutdown();}if (this.putMessageFutureExecutor != null) {this.putMessageFutureExecutor.shutdown();}if (this.pullMessageExecutor != null) {this.pullMessageExecutor.shutdown();}if (this.replyMessageExecutor != null) {this.replyMessageExecutor.shutdown();}if (this.adminBrokerExecutor != null) {this.adminBrokerExecutor.shutdown();}if (this.brokerOuterAPI != null) {this.brokerOuterAPI.shutdown();}this.consumerOffsetManager.persist();if (this.filterServerManager != null) {this.filterServerManager.shutdown();}if (this.brokerFastFailure != null) {this.brokerFastFailure.shutdown();}if (this.consumerFilterManager != null) {this.consumerFilterManager.persist();}if (this.clientManageExecutor != null) {this.clientManageExecutor.shutdown();}if (this.queryMessageExecutor != null) {this.queryMessageExecutor.shutdown();}if (this.consumerManageExecutor != null) {this.consumerManageExecutor.shutdown();}if (this.fileWatchService != null) {this.fileWatchService.shutdown();}if (this.transactionalMessageCheckService != null) {this.transactionalMessageCheckService.shutdown(false);}if (this.endTransactionExecutor != null) {this.endTransactionExecutor.shutdown();}}

8. 小结

在本篇文章中，我们对 broker 启动的源码进行了简要梳理，着重介绍了 broker 启动过程中所涉及的各项服务，以及启动流程中具体执行的操作。至于这些服务的具体功能，以及其背后源码的详细实现，我们将留待下一篇文章展开深入探讨。

如有错误，欢迎指出！！！！