RocketMQ broker启动流程是什么

这篇文章主要介绍“RocketMQ broker启动流程是什么”的相关知识，小编通过实际案例向大家展示操作过程，操作方法简单快捷，实用性强，希望这篇“RocketMQ broker启动流程是什么”文章能帮助大家解决问题。

1. 启动入口

本系列RocketMQ4.8注释GitHub地址，希望对大家有所帮助，要是觉得可以的话麻烦给点一下Star哈

前面我们已经分析完了NameServer和producer，从本文开始，我们将分析Broker。

broker的启动类为org.apache.rocketmq.broker.BrokerStartup，代码如下：

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

在main()方法中，仅有一行代码，这行代码包含了两个操作:

• createBrokerController(...)：创建BrokerController

• start(...)：启动Broker

接下来我们就来分析这两个操作。

2. 创建BrokerController

创建BrokerController的方法为BrokerStartup#createBrokerController，代码如下：

public static BrokerController createBrokerController(String[] args) {    ...    try {        //解析命令行参数        Options options = ServerUtil.buildCommandlineOptions(new Options());        commandLine = ServerUtil.parseCmdLine("mqbroker", args, buildCommandlineOptions(options),            new PosixParser());        if (null == commandLine) {            System.exit(-1);        }        // 处理配置        final BrokerConfig brokerConfig = new BrokerConfig();        final NettyServerConfig nettyServerConfig = new NettyServerConfig();        final NettyClientConfig nettyClientConfig = new NettyClientConfig();        // tls安全相关        nettyClientConfig.setUseTLS(Boolean.parseBoolean(System.getProperty(TLS_ENABLE,            String.valueOf(TlsSystemConfig.tlsMode == TlsMode.ENFORCING))));        // 配置端口        nettyServerConfig.setListenPort(10911);        // 消息存储的配置        final MessageStoreConfig messageStoreConfig = new MessageStoreConfig();        ...        // 将命令行中的配置设置到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);        }        //省略一些配置        ...        // 创建 brokerController        final BrokerController controller = new BrokerController(            brokerConfig,            nettyServerConfig,            nettyClientConfig,            messageStoreConfig);        controller.getConfiguration().reGISterConfig(properties);        // 初始化        boolean initResult = controller.initialize();        if (!initResult) {            controller.shutdown();            System.exit(-3);        }        // 关闭钩子，在关闭前处理一些操作        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {            private volatile boolean hasshutdown = false;            private AtomicInteger shutdownTimes = new AtomicInteger(0);            @Override            public void run() {                synchronized (this) {                    if (!this.hasShutdown) {                        ...                        // 这里会发一条注销消息给nameServer                        controller.shutdown();                        ...                    }                }            }        }, "ShutdownHook"));        return controller;    } catch (Throwable e) {        e.printStackTrace();        System.exit(-1);    }    return null;}

这个方法的代码有点长，但功能并不多，总的来说就三个功能：

• 处理配置：主要是处理nettyServerConfig与nettyClientConfig的配置，这块就是一些配置解析的操作，处理方式与NameServer很类似，这里就不多说了。

• 创建及初始化controller：调用方法controller.initialize()，这块内容我们后面分析。

• 注册关闭钩子：调用Runtime.getRuntime().addShutdownHook(...)，可以在JVM进程关闭前进行一些操作。

2.1 controller实例化

BrokerController的创建及初始化是在BrokerStartup#createBrokerController方法中进行，我们先来看看它的构造方法：

public BrokerController(    final BrokerConfig brokerConfig,    final NettyServerConfig nettyServerConfig,    final NettyClientConfig nettyClientConfig,    final MessageStoreConfig messageStoreConfig) {    // 4个核心配置信息    this.brokerConfig = brokerConfig;    this.nettyServerConfig = nettyServerConfig;    this.nettyClientConfig = nettyClientConfig;    this.messageStoreConfig = messageStoreConfig;    // 管理consumer消费消息的offset    this.consumerOffsetManager = new ConsumerOffsetManager(this);    // 管理topic配置    this.topicConfigManager = new TopicConfigManager(this);    // 处理 consumer 拉消息请求的    this.pullMessageProcessor = new PullMessageProcessor(this);    this.pullRequestHoldService = new PullRequestHoldService(this);    // 消息送达的监听器    this.messageArrivingListener         = new NotifyMessageArrivingListener(this.pullRequestHoldService);    ...    // 往外发消息的组件    this.brokerOuterapi = new BrokerOuterAPI(nettyClientConfig);    ...}

BrokerController的构造方法很长，基本都是一些赋值操作，代码中已列出关键项，这些包括：

• 核心配置赋值：主要是brokerConfig/nettyServerConfig/nettyClientConfig/messageStoreConfig四个配置

• ConsumerOffsetManager：管理consumer消费消息位置的偏移量，偏移量表示消费者组消费该topic消息的位置，后面再消费时，就从该位置后消费，避免重复消费消息，也避免了漏消费消息。

• topicConfigManager：topic配置管理器，就是用来管理topic配置的，如topic名称，topic队列数量

• pullMessageProcessor：消息处理器，用来处理消费者拉消息

• messageArrivingListener：消息送达的监听器，当生产者的消息送达时，由该监听器监听

• brokerOuterAPI：往外发消息的组件，如向NameServer发送注册/注销消息

以上这些组件的用处，这里先混个脸熟，我们后面再分析。

2.2 初始化controller

我们再来看看初始化操作，方法为BrokerController#initialize：

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);            // 启用了DLeger,就创建DLeger相关组件            if (messageStoreConfig.isEnableDLegerCommitLog()) {                ...            }            // 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);            this.messageStore.getDispatcherList().addFirst(                new CommitLogDispatcherCalcBitMap(this.brokerConfig, this.consumerFilterManager));        } catch (IOException e) {            result = false;            log.error("Failed to initialize", e);        }    }    // 加载磁盘上的记录，如commitLog写入的位置、消费者主题/队列的信息    result = result && this.messageStore.load();    if (result) {        // 处理 nettyServer        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);        // 创建线程池start... 这里会创建多种类型的线程池        ...        // 处理consumer pull操作的线程池        this.pullMessageExecutor = new BrokerFixedThreadPoolExecutor(            this.brokerConfig.getPullMessageThreadPoolNums(),            this.brokerConfig.getPullMessageThreadPoolNums(),            1000 * 60,            TimeUnit.MILLISECONDS,            this.pullThreadPoolQueue,            new ThreadFactoryImpl("PullMessageThread_"));        ...        // 创建线程池end...        // 注册处理器        this.registerProcessor();        // 启动定时任务start... 这里会启动好多的定时任务        ...        // 定时将consumer消费到的offset进行持久化操作，即将数据保存到磁盘上        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);        ...        // 启动定时任务end...        ...        // 开启 DLeger 的一些操作        if (!messageStoreConfig.isEnableDLegerCommitLog()) {            ...        }        // 处理tls配置        if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {            ...        }        // 初始化一些操作        initialTransaction();        initialAcl();        initialrpcHooks();    }    return result;}

这个还是很长，关键部分都做了注释，该方法所做的工作如下：

• 加载配置文件中的配置

• 赋值与初始化操作

• 创建线程池

• 注册处理器

• 启动定时任务

这里我们来看下注册处理器的操作this.registerProcessor():

2.2.1 注册处理器：BrokerController#registerProcessor

this.registerProcessor()实际调用的方法是BrokerController#registerProcessor，代码如下：

public void registerProcessor() {        SendMessageProcessor sendProcessor = new SendMessageProcessor(this);    sendProcessor.registerSendMessageHook(sendMessageHookList);    sendProcessor.registerConsumeMessageHook(consumeMessageHookList);    this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor,         this.sendMessageExecutor);    this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor,          this.sendMessageExecutor);    this.remotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendProcessor,         this.sendMessageExecutor);    this.remotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor,         this.sendMessageExecutor);    ...        this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE, this.pullMessageProcessor,         this.pullMessageExecutor);    this.pullMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);        ReplyMessageProcessor replyMessageProcessor = new ReplyMessageProcessor(this);    replyMessageProcessor.registerSendMessageHook(sendMessageHookList);    ...}

这个方法里注册了许许多多的处理器，这里仅列出了与消息相关的内容，如发送消息、回复消息、拉取消息等，后面在处理producer/consumer的消息时，就会用到这些处理器，这里先不展开分析。

2.2.2 remotingServer注册处理器：NettyRemotingServer#registerProcessor

我们来看下remotingServer注册处理器的操作，方法为NettyRemotingServer#registerProcessor：

public class NettyRemotingServer extends NettyRemotingAbstract implements RemotingServer {    ...    @Override    public void registerProcessor(int requestCode, NettyRequestProcessor processor,             ExecutorService executor) {        ExecutorService executorThis = executor;        if (null == executor) {            executorThis = this.publicExecutor;        }        Pair<NettyRequestProcessor, ExecutorService> pair = new Pair<NettyRequestProcessor,                 ExecutorService>(processor, executorThis);        // 注册到processorTable 中        this.processorTable.put(requestCode, pair);    }    ...}

最终，这些处理器注册到了processorTable中，它是NettyRemotingAbstract的成员变量，定义如下：

HashMap<Integer, Pair<NettyRequestProcessor, ExecutorService>>

这是一个hashMap的结构，key为code，value为Pair，该类中有两个成员变量：NettyRequestProcessor、ExecutorService，code与NettyRequestProcessor的映射关系就是在hashMap里存储的。

2.3 注册关闭钩子：Runtime.getRuntime().addShutdownHook(...)

接着我们来看看注册关闭钩子的操作：

// 关闭钩子，在关闭前处理一些操作Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {    private volatile boolean hasShutdown = false;    private AtomicInteger shutdownTimes = new AtomicInteger(0);    @Override    public void run() {        synchronized (this) {            if (!this.hasShutdown) {                ...                // 这里会发一条注销消息给nameServer                controller.shutdown();                ...            }        }    }}, "ShutdownHook"));

跟进BrokerController#shutdown方法：

public void shutdown() {    // 调用各组件的shutdown方法    ...    // 发送注销消息到NameServer    this.unregisterBrokerAll();    ...    // 持久化consumer的消费偏移量    this.consumerOffsetManager.persist();    // 又是调用各组件的shutdown方法    ...

这个方法里会调用各组件的shutdown()方法、发送注销消息给NameServer、持久化consumer的消费偏移量，这里我们主要看发送注销消息的方法BrokerController#unregisterBrokerAll:

private void unregisterBrokerAll() {    // 发送一条注销消息给nameServer    this.brokerOuterAPI.unregisterBrokerAll(        this.brokerConfig.getBrokerClusterName(),        this.getBrokerAddr(),        this.brokerConfig.getBrokerName(),        this.brokerConfig.getBrokerId());}

继续进入BrokerOuterAPI#unregisterBrokerAll：

public void unregisterBrokerAll(    final String clusterName,    final String brokerAddr,    final String brokerName,    final long brokerId) {    // 获取所有的 nameServer，遍历发送注销消息    List<String> nameServerAddressList = this.remotinGClient.getNameServerAddressList();    if (nameServerAddressList != null) {        for (String namesrvAddr : nameServerAddressList) {            try {                this.unregisterBroker(namesrvAddr, clusterName, brokerAddr, brokerName, brokerId);                log.info("unregisterBroker OK, NamesrvAddr: {}", namesrvAddr);            } catch (Exception e) {                log.warn("unregisterBroker Exception, {}", namesrvAddr, e);            }        }    }}

这个方法里，会获取到所有的nameServer，然后逐个发送注销消息，继续进入BrokerOuterAPI#unregisterBroker方法：

public void unregisterBroker(    final String namesrvAddr,    final String clusterName,    final String brokerAddr,    final String brokerName,    final long brokerId) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,         InterruptedException, MQBrokerException {    UnRegisterBrokerRequestHeader requestHeader = new UnRegisterBrokerRequestHeader();    requestHeader.setBrokerAddr(brokerAddr);    requestHeader.setBrokerId(brokerId);    requestHeader.setBrokerName(brokerName);    requestHeader.setClusterName(clusterName);    // 发送的注销消息：RequestCode.UNREGISTER_BROKER    RemotingCommand request = RemotingCommand.createRequestCommand(            c, requestHeader);    RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, 3000);    assert response != null;    switch (response.getCode()) {        case ResponseCode.SUCCESS: {            return;        }        default:            break;    }    throw new MQBrokerException(response.getCode(), response.getRemark(), brokerAddr);}

最终调用的是RemotingClient#invokeSync进行消息发送，请求code是RequestCode.UNREGISTER_BROKER，这就与NameServer接收broker的注销消息对应上了。

3. 启动Broker：start(...)

我们再来看看Broker的启动流程，处理方法为BrokerController#start：

public void start() throws Exception {    // 启动各组件    // 启动消息存储相关组件    if (this.messageStore != null) {        this.messageStore.start();    }    // 启动 remotingServer，其实就是启动一个netty服务，用来接收producer传来的消息    if (this.remotingServer != null) {        this.remotingServer.start();    }    ...    // broker对外发放消息的组件，向nameServer上报存活消息时使用了它，也是一个netty服务    if (this.brokerOuterAPI != null) {        this.brokerOuterAPI.start();    }    ...    // broker 核心的心跳注册任务    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {        @Override        public void run() {            try {                BrokerController.this.registerBrokerAll(true, false,                     brokerConfig.isForceRegister());            } catch (Throwable e) {                log.error("registerBrokerAll Exception", e);            }        }        // brokerConfig.getRegisterNameServerPeriod() 值为 1000 * 30，最终计算得到默认30秒执行一次    }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)),             TimeUnit.MILLISECONDS);    ...}

这个方法主要就是启动各组件了，这里列出了几大重要组件的启动：

• messageStore：消息存储组件，在这个组件里，会启动消息存储相关的线程，如消息的投递操作、commitLog文件的flush操作、comsumeQueue文件的flush操作等

• remotingServer：netty服务，用来接收请求消息，如producer发送过来的消息

• brokerOuterAPI：也是一个netty服务，用来对外发送消息，如向nameServer上报心跳消息

• 启动定时任务：broker向nameServer发送注册消息

这里我们重点来看定时任务是如何发送心跳发送的。

处理注册消息发送的时间间隔如下：

Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)

这行代码看着长，但意思就一句话：时间间隔可以自行配置，但不能小于10s，不能大于60s，默认是30s.

处理消息注册的方法为BrokerController#registerBrokerAll(...)，代码如下：

public synchronized void registerBrokerAll(final boolean checkOrderConfig,         boolean oneway, boolean forceRegister) {    TopicConfigSerializeWrapper topicConfigWrapper             = this.getTopicConfigManager().buildTopicConfigSerializeWrapper();    // 处理topic相关配置    if (!PermName.isWriteable(this.getBrokerConfig().getBrokerPermission())        || !PermName.isReadable(this.getBrokerConfig().getBrokerPermission())) {        ...    }    // 这里会判断是否需要进行注册    if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),        this.getBrokerAddr(),        this.brokerConfig.getBrokerName(),        this.brokerConfig.getBrokerId(),        this.brokerConfig.getRegisterBrokerTimeoutMills())) {        // 进行注册操作            doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);    }}

这个方法就是用来处理注册操作的，不过注册前会先验证下是否需要注册，验证是否需要注册的方法为BrokerController#needRegister, 代码如下：

private boolean needRegister(final String clusterName,    final String brokerAddr,    final String brokerName,    final long brokerId,    final int timeoutMills) {    TopicConfigSerializeWrapper topicConfigWrapper         = this.getTopicConfigManager().buildTopicConfigSerializeWrapper();    // 判断是否需要进行注册    List<Boolean> changeList = brokerOuterAPI.needRegister(clusterName, brokerAddr, brokerName,         brokerId, topicConfigWrapper, timeoutMills);    // 有一个发生了变化，就表示需要注册了        boolean needRegister = false;    for (Boolean changed : changeList) {        if (changed) {            needRegister = true;            break;        }    }    return needRegister;}

这个方法调用了brokerOuterAPI.needRegister(...)来判断broker是否发生了变化，只要一个NameServer上发生了变化，就说明需要进行注册操作。

brokerOuterAPI.needRegister(...)是如何判断broker是否发生了变化的呢？继续跟进BrokerOuterAPI#needRegister：

public List<Boolean> needRegister(    final String clusterName,    final String brokerAddr,    final String brokerName,    final long brokerId,    final TopicConfigSerializeWrapper topicConfigWrapper,    final int timeoutMills) {    final List<Boolean> changedList = new CopyOnWriteArrayList<>();    // 获取所有的 nameServer    List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();    if (nameServerAddressList != null && nameServerAddressList.size() > 0) {        final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());        // 遍历所有的nameServer,逐一发送请求        for (final String namesrvAddr : nameServerAddressList) {            brokerOuterExecutor.execute(new Runnable() {                @Override                public void run() {                    try {                        QueryDataVersionRequestHeader requestHeader                             = new QueryDataVersionRequestHeader();                        ...                        // 向nameServer发送消息，命令是 RequestCode.QUERY_DATA_VERSION                        RemotingCommand request = RemotingCommand                            .createRequestCommand(RequestCode.QUERY_DATA_VERSION, requestHeader);                        // 把当前的 DataVersion 发到 nameServer                             request.setBody(topicConfigWrapper.getDataVersion().encode());                        // 发请求到nameServer                        RemotingCommand response = remotingClient                            .invokeSync(namesrvAddr, request, timeoutMills);                        DataVersion nameServerDataVersion = null;                        Boolean changed = false;                        switch (response.getCode()) {                            case ResponseCode.SUCCESS: {                                QueryDataVersionResponseHeader queryDataVersionResponseHeader =                                  (QueryDataVersionResponseHeader) response                                  .decodeCommandCustomHeader(QueryDataVersionResponseHeader.class);                                changed = queryDataVersionResponseHeader.getChanged();                                byte[] body = response.getBody();                                if (body != null) {                                    // 拿到 DataVersion                                    nameServerDataVersion = DataVersion.decode(body, D                                        ataVersion.class);                                    // 这里是判断的关键                                    if (!topicConfigWrapper.getDataVersion()                                        .equals(nameServerDataVersion)) {                                        changed = true;                                    }                                }                                if (changed == null || changed) {                                    changedList.add(Boolean.TRUE);                                }                            }                            default:                                break;                        }                        ...                    } catch (Exception e) {                        ...                    } finally {                        countDownLatch.countDown();                    }                }            });        }        try {            countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);        } catch (InterruptedException e) {            log.error("query dataversion from nameserver countDownLatch await Exception", e);        }    }    return changedList;}

这个方法里，先是遍历所有的nameServer，向每个nameServer都发送一条code为RequestCode.QUERY_DATA_VERSION的参数，参数为当前broker的DataVersion，当nameServer收到消息后，就返回nameServer中保存的、与当前broker对应的DataVersion，当两者版本不相等时，就表明当前broker发生了变化，需要重新注册。

DataVersion是个啥呢？它的部分代码如下：

public class DataVersion extends RemotingSerializable {    // 时间戳    private long timestamp = System.currentTimeMillis();    // 计数器，可以理解为最近的版本号    private AtomicLong counter = new AtomicLong(0);    public void nextVersion() {        this.timestamp = System.currentTimeMillis();        this.counter.incrementAndGet();    }        @Override    public boolean equals(final Object o) {        if (this == o)            return true;        if (o == null || getClass() != o.getClass())            return false;        final DataVersion that = (DataVersion) o;        if (timestamp != that.timestamp) {            return false;        }        if (counter != null && that.counter != null) {            return counter.longValue() == that.counter.longValue();        }        return (null == counter) && (null == that.counter);    }    ...}

从DataVersion的equals()方法来看，只有当timestamp与counter都相等时，两个DataVersion对象才相等。那这两个值会在哪里被修改呢？从DataVersion#nextVersion方法的调用情况来看，引起这两个值的变化主要有两种：

• broker 上新创建了一个 topic

• topic的发了的变化

在这两种情况下，DataVersion#nextVersion方法被调用，从而引起DataVersion的改变。DataVersion改变了，就表明当前broker需要向nameServer注册了。

让我们再回到BrokerController#registerBrokerAll(...)方法：

public synchronized void registerBrokerAll(final boolean checkOrderConfig,         boolean oneway, boolean forceRegister) {    ...    // 这里会判断是否需要进行注册    if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),        this.getBrokerAddr(),        this.brokerConfig.getBrokerName(),        this.brokerConfig.getBrokerId(),        this.brokerConfig.getRegisterBrokerTimeoutMills())) {        // 进行注册操作            doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);    }}

处理注册的方法为BrokerController#doRegisterBrokerAll，稍微看下它的流程：

private void doRegisterBrokerAll(boolean checkOrderConfig, boolean oneway,        TopicConfigSerializeWrapper topicConfigWrapper) {    // 注册    List<RegisterBrokerResult> registerBrokerResultList = this.brokerOuterAPI.registerBrokerAll(        this.brokerConfig.getBrokerClusterName(),        this.getBrokerAddr(),        this.brokerConfig.getBrokerName(),        this.brokerConfig.getBrokerId(),        this.getHAServerAddr(),        // 这个对象里就包含了当前broker的版本信息        topicConfigWrapper,        this.filterServerManager.buildNewFilterServerList(),        oneway,        this.brokerConfig.getRegisterBrokerTimeoutMills(),        this.brokerConfig.isCompressedRegister());    ...}

继续跟下去，最终调用的是BrokerOuterAPI#registerBroker方法：

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 request = RemotingCommand        .createRequestCommand(RequestCode.REGISTER_BROKER, requestHeader);    request.setBody(body);    // 处理发送操作：sendOneWay    if (oneway) {        try {            // 注册操作            this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);        } catch (RemotingTooMuchRequestException e) {            // Ignore        }        return null;        ...    }    ....}

所以，所谓的注册操作，就是当nameServer发送一条code为RequestCode.REGISTER_BROKER的消息，消息里会带上当前broker的topic信息、版本号等。

关于“RocketMQ broker启动流程是什么”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识，可以关注编程网精选频道，小编每天都会为大家更新不同的知识点。