开发者

RocketMQ源码解析broker 启动流程

目录
  • 1. 启动入口
  • 2. 创建BrokerController
    • 2.1 controller实例化
    • 2.2 初始化controller
      • 2.2.1 注册处理器:BrokerController#registerProcessor
      • 2.2.2 remotingServer注册处理器:NettyRemotingServer#registerProcessor
    • 2.3 注册关闭钩子:Runtime.getRuntime().addShutdownHook(...)
    • 3. 启动Broker:start(...)
      • 4.总结

        1. 启动入口

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

        前面我们已经分析完了NameServerproducer,从本文开始,我们将分析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,代码如下:

        /**
         * 创建 broker 的配置参数
         */
        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.ejavascriptxit(-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;
        }
        

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

        • 处理配置:主要是处理nettyServerConfignettyClientConfig的配置,这块就是一些配置解析的操作,处理方式与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消息的位置,后面再消费时,就从该位置后消费,避免重复消费消息,也避免了漏消费消息。
        • topicConfigManagertopic配置管理器,就是用来管理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();
         开发者_C入门   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... 这里会启动好多的定时任务
                ...
                // 定时将cojsnsumer消费到的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
             */
            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);
            ...
            /**
             * PullMessageProcessor
             */
            this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE, this.pullMessageProcessor, 
                this.pullMessageExecutor);
            this.pullMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);
            /**
                * ReplyMessageProcessor
                */
            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/* request code */, Pair<NettyRequestProcessor, ExecutorService>>
        

        这是一个hashMap的结构,keycodevaluePair,该类中有两个成员变量:NettyRequestProcessorExecutorServicecodeNettyRequestProcessor的映射关系就是在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();
                        ...
           js         }
                }
            }
        }, "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进行消息发送,请求codeRequestCode.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操作等
        • remotingServernetty服务,用来接收请求消息,如producer发送过来的消息
        • brokerOuterAPI:也是一个netty服务,用来对外发送消息,如向nameServer上报心跳消息
        • 启动定时任务:brokernameServer发送注册消息

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

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

        Math.max(10000, Math.min(brokerConfig.getRegisterNameSwww.devze.comerverPeriod(), 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.getBrokerConyjpiiDLqhfig().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&lt;Boolean&gt; 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都发送一条codeRequestCode.QUERY_DATA_VERSION的参数,参数为当前brokerDataVersion,当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();
            }
            /**
             * equals 方法,当 timestamp 与 counter 都相等时,则两者相等
             */
            @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);
            }
            ...
        } 
        

        DataVersionequals()方法来看,只有当timestampcounter都相等时,两个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发送一条codeRequestCode.REGISTER_BROKER的消息,消息里会带上当前brokertopic信息、版本号等。

        4.总结

        本文主要分析了broker的启动流程,总的来说,启动流程分为3个:

        • 解析配置文件,这一步会解析各种配置,并将其赋值到对应的对象中
        • BrokerController创建及初始化:创建了BrokerController对象,并进行初始化操作,所谓的初始化,就是加载配置文件中配置、创建线程池、注册请求处理器、启动定时任务等
        • BrokerController启动:这一步是启动broker的核心组件,如messageStore(消息存储)、remotingServer(netty服务,用来处理producerconsumer请求)、brokerOuterAPI(netty服务,用来向nameServer上报当前broker信息)等。

        在分析启动过程中,重点分析了两类消息的发送:

        • ShutdownHook中,broker会向nameServer发送注销消息,这表明在broker关闭前,nameServer会清除当前broker的注册信息

        • broker启动后,会启动一个定时任务,定期判断是否需要向nameServer注册,判断是否需要注册时,会向nameServer发送codeQUERY_DATA_VERSION的消息,从nameServer得到当前broker的版本号,该版本号与本地版本号不一致,就表示需要向broker重新注册了,即发送注册消息。

        参考文章

        RocketMQ4.8注释github地址

        以上就是RocketMQ源码解析broker 启动流程的详细内容,更多关于RocketMQ broker启动的资料请关注我们其它相关文章!

        0

        上一篇:

        下一篇:

        精彩评论

        暂无评论...
        验证码 换一张
        取 消

        最新开发

        开发排行榜