RocketMQ源码解析——Broker部分之Broker启动过

作者: szhlcy | 来源:发表于2021-08-03 21:05 被阅读0次

@[toc]

从启动脚本到启动类

我们知道RocketMQ的Broker端的启动方式为，进入到RocketMQ的bin目录下，运行对应的mqbroker脚本。
RocketMQ的Linux和Windows脚本内容也就是对应的mqbroker脚本，这里贴出来看看

//省略部分脚本
export ROCKETMQ_HOME

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

上面这段是Linux系统下的shell脚本，

if not exist "%ROCKETMQ_HOME%\bin\runbroker.cmd" echo Please set the ROCKETMQ_HOME variable in your environment! & EXIT /B 1

call "%ROCKETMQ_HOME%\bin\runbroker.cmd" org.apache.rocketmq.broker.BrokerStartup %*

IF %ERRORLEVEL% EQU 0 (
   ECHO "Broker starts OK"
)

上面这段是Windows系统下的cmd脚本命令。从这两段脚本中可以看出，最后的都会运行一个runbroker的脚本，还会把一个类的全路径名作为参数传进去，这个类就是启动类BrokerStartup。

启动类`BrokerStartup`

BrokerStartup的启动函数是main函数，里面的代码也简单只有一行，主要的逻辑也简单，就是创建BrokerController类（后面说），然后启动。

    public static void main(String[] args) {
        //创建createBrokerController然后启动
        start(createBrokerController(args));
    }

启动的准备工作`createBrokerController`

public static BrokerController createBrokerController(String[] args) {
        //MQ的版本号
        System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
        //设置broker的netty客户端的发送缓冲大小，默认是128 kb
        if (null == System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_SNDBUF_SIZE)) {
            NettySystemConfig.socketSndbufSize = 131072;
        }
        //设置broker的netty客户端的接收缓冲大小，默认是128 kb
        if (null == System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_RCVBUF_SIZE)) {
            NettySystemConfig.socketRcvbufSize = 131072;
        }

        try {
            //PackageConflictDetect.detectFastjson();
            //命令行选项解析
            Options options = ServerUtil.buildCommandlineOptions(new Options());
            //解析命令栏中的 mqbroker
            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 （TLS是SSL的升级版本，TLS是SSL的标准化后的产物，有1.0 1.1 1.2三个版本）
            nettyClientConfig.setUseTLS(Boolean.parseBoolean(System.getProperty(TLS_ENABLE,
                String.valueOf(TlsSystemConfig.tlsMode == TlsMode.ENFORCING))));
            //设置netty的服务端监听的端口 10911
            nettyServerConfig.setListenPort(10911);
            //消息存储相关的配置
            final MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
            //如果Broker是Slave角色，消息占用内存大小的比率 比默认的40% 还要小 10%
            if (BrokerRole.SLAVE == messageStoreConfig.getBrokerRole()) {
                int ratio = messageStoreConfig.getAccessMessageInMemoryMaxRatio() - 10;
                //设置消息的内存最大占比，如果内存占比超过设定值，那么就进行置换
                messageStoreConfig.setAccessMessageInMemoryMaxRatio(ratio);
            }


           ......省略配置读取保存相关代码


            //创建BrokerController
            final BrokerController controller = new BrokerController(
                brokerConfig,
                nettyServerConfig,
                nettyClientConfig,
                messageStoreConfig);
            // remember all configs to prevent discard
            //缓存额外配置
            controller.getConfiguration().registerConfig(properties);
            //初始化BrokerController
            boolean initResult = controller.initialize();
            if (!initResult) {
                controller.shutdown();
                System.exit(-3);
            }
            //注册关闭的钩子方法
            Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
                private volatile boolean hasShutdown = false;
                private AtomicInteger shutdownTimes = new AtomicInteger(0);
                @Override
                public void run() {
                    synchronized (this) {
                        log.info("Shutdown hook was invoked, {}", this.shutdownTimes.incrementAndGet());
                        if (!this.hasShutdown) {
                            this.hasShutdown = true;
                            long beginTime = System.currentTimeMillis();
                            //BrokerController的销毁方法
                            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;
    }

createBrokerController这个方法逻辑比较简单，主要就是获取启动参数和命令单中的配置参数以及默认的配置参数进行融合。然后使用这些配置对象创建BrokerController，并调用其initialize方法。其中核心就是创建BrokerController，并调用其initialize方法。这里主要说一下Master和Slave在启动时候的区别

参数	说明	master	slave
`accessMessageInMemoryMaxRatio`	消息占用内存大小的比率，超过这个值就需要进行内存置换	默认值40%	默认值30%
`brokerId`	机器的角色Id	0	大于0的数

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

端口	作用
10911	接收消息推送的端口
10912	高可用的端口
10909	推送消息的vip端口

Broker端的核心`BrokerController`

`BrokerController`的构造方法

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

public BrokerController(
        final BrokerConfig brokerConfig,
        final NettyServerConfig nettyServerConfig,
        final NettyClientConfig nettyClientConfig,
        final MessageStoreConfig messageStoreConfig
    ) {
        //前面准备的配置信息
        this.brokerConfig = brokerConfig;
        this.nettyServerConfig = nettyServerConfig;
        this.nettyClientConfig = nettyClientConfig;
        this.messageStoreConfig = messageStoreConfig;
        //Consumer消费进度记录管理类
        this.consumerOffsetManager = new ConsumerOffsetManager(this);

        //消息topic维度的管理查询类 管理topic和topic相关配置关系
        this.topicConfigManager = new TopicConfigManager(this);

        //Consumer端使用pull的方式向Broker拉取消息请求的处理类
        this.pullMessageProcessor = new PullMessageProcessor(this);

        //Consumer端使用pull的方式拉取请求时，保存请求，当有消息到达时进行推送处理类
        this.pullRequestHoldService = new PullRequestHoldService(this);

        //消息到达Broker的时候的监听回调类，这里会调用到pullRequestHoldService中的方法
        this.messageArrivingListener = new NotifyMessageArrivingListener(this.pullRequestHoldService);

        //消费者id变化监听器
        this.consumerIdsChangeListener = new DefaultConsumerIdsChangeListener(this);

        //消费者管理类  按照group进行分组，对消费者的id变化进行监听
        this.consumerManager = new ConsumerManager(this.consumerIdsChangeListener);

        //消费者的过滤管理类  按照topic进行分类
        this.consumerFilterManager = new ConsumerFilterManager(this);

        //生产者管理 按照group进行分类
        this.producerManager = new ProducerManager();

        //心跳连接处理类
        this.clientHousekeepingService = new ClientHousekeepingService(this);

        //控制台用的
        this.broker2Client = new Broker2Client(this);

        //订阅关系管理类
        this.subscriptionGroupManager = new SubscriptionGroupManager(this);

        //broker对外api
        this.brokerOuterAPI = new BrokerOuterAPI(nettyClientConfig);

        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
        );
    }

列举一下这些核心类的作用

字段类	作用
`ConsumerOffsetManager`	Consumer消费者的消费进度记录管理类
`TopicConfigManager`	消息topic维度的管理查询类
`PullMessageProcessor`	Consumer端使用pull的方式向Broker拉取消息请求的处理类
`PullRequestHoldService`	Consumer端使用push的方式拉取请求时，保存请求，当有消息到达时进行推送处理类。对于push方式消费，会对消费端的请求进行保存，当有消息到达的时候然后进行推送
`NotifyMessageArrivingListener`	消息到达Broker的时候的监听回调类，这里会调用到pullRequestHoldService中的notifyMessageArriving方法
`DefaultConsumerIdsChangeListener`	消费者id变化监听器,主要监听Consumer的注册和下线的事件
`ConsumerManager`	消费者管理类按照group进行分组，对消费者的id变化进行监听
`ConsumerFilterManager`	消费者的过滤管理类按照topic进行分类
`ProducerManager`	生产者管理按照group进行分类
`ClientHousekeepingService`	心跳连接处理类
`Broker2Client`	控制台获取Broker信息用
`SubscriptionGroupManager`	订阅关系管理类
`BrokerOuterAPI`	broker对外api
`SlaveSynchronize`	从broker 同步进度管理类

`BrokerController`的初始化`initialize`

在BrokerStartup的createBrokerController方法中，在创建了BrokerController对象之后会马上调用其initialize方法。来进行Broker的初始化。
initialize方法，内部的逻辑比较多。这里整理一下主要做的事情：

从服务器加载topic配置，不同的Consumer消费的进度情况，订阅关系，Consumer的过滤信息配置等信息。这些信息都保存在对应的文件当中
如果步骤1加载正常，会创建DefaultMessageStore对象。在这个对象中又有一堆逻辑

public boolean initialize() throws CloneNotSupportedException {
        //加载 topic 相关配置，文件地址为 {user.home}/store/config/topics.json
        boolean result = this.topicConfigManager.load();
        //加载 不同的Consumer消费的进度情况  文件地址为 {user.home}/store/config/consumerOffset.json
        result = result && this.consumerOffsetManager.load();
        //加载 订阅关系  文件地址  {user.home}/store/config/subscriptionGroup.json
        result = result && this.subscriptionGroupManager.load();
        //加载 Consumer的过滤信息配置  文件地址  {user.home}/store/config/consumerFilter.json
        result = result && this.consumerFilterManager.load();
        //如果上述文件加载正常
        if (result) {
            try {
                //创建DefaultMessageStore，
                this.messageStore =
                    new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener,
                        this.brokerConfig);
                //使用的是DLegerCommitLog，则创建DLedgerRoleChangeHandler
                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);
                this.messageStore.getDispatcherList().addFirst(new CommitLogDispatcherCalcBitMap(this.brokerConfig, this.consumerFilterManager));
            } catch (IOException e) {
                result = false;
                log.error("Failed to initialize", e);
            }
        }
        //加载消息的日志文件，包含CommitLog，ConsumeQueue等
        result = result && this.messageStore.load();
        //如果上述文件存在一个加载不成功则直接返回
        if (result) {
            //开启服务端
            this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.clientHousekeepingService);
            NettyServerConfig fastConfig = (NettyServerConfig) this.nettyServerConfig.clone();
            fastConfig.setListenPort(nettyServerConfig.getListenPort() - 2);
            //再开一个端口为10909 的服务端口，这个端口只给 消息的生产者使用
            this.fastRemotingServer = new NettyRemotingServer(fastConfig, this.clientHousekeepingService);
            //处理消息生产者发送的生成消息api 相关的线程池
            this.sendMessageExecutor = new BrokerFixedThreadPoolExecutor(
                this.brokerConfig.getSendMessageThreadPoolNums(),
                this.brokerConfig.getSendMessageThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.sendThreadPoolQueue,
                new ThreadFactoryImpl("SendMessageThread_"));
            //处理消费者发出的消费消息api 相关的线程池
            this.pullMessageExecutor = new BrokerFixedThreadPoolExecutor(
                this.brokerConfig.getPullMessageThreadPoolNums(),
                this.brokerConfig.getPullMessageThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.pullThreadPoolQueue,
                new ThreadFactoryImpl("PullMessageThread_"));
            //处理回复消息api的线程池
            this.replyMessageExecutor = new BrokerFixedThreadPoolExecutor(
                this.brokerConfig.getProcessReplyMessageThreadPoolNums(),
                this.brokerConfig.getProcessReplyMessageThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.replyThreadPoolQueue,
                new ThreadFactoryImpl("ProcessReplyMessageThread_"));
            //查询线程
            this.queryMessageExecutor = new BrokerFixedThreadPoolExecutor(
                this.brokerConfig.getQueryMessageThreadPoolNums(),
                this.brokerConfig.getQueryMessageThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.queryThreadPoolQueue,
                new ThreadFactoryImpl("QueryMessageThread_"));
            //-------------------------省略部分线程池的创建------------------------------//
            
            //为客户端注册需要处理API指令事件，以及消息发送和消费的回调方法
            this.registerProcessor();
            
            //-------------------------省略定时执行任务服务------------------------------//
            //初始化事务消息相关的服务
            initialTransaction();
            //消息轨迹
            initialAcl();
            //Rpc调用的钩子
            initialRpcHooks();
    }


    private void initialTransaction() {
        //加载TransactionalMessageService服务，实现类为TransactionalMessageServiceImpl
        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());
        }
        //AbstractTransactionalMessageCheckListener对应的服务类为LogTransactionalMessageCheckListener ，其中实现为空实现
        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到AbstractTransactionalMessageCheckListener中
        this.transactionalMessageCheckListener.setBrokerController(this);
        //创建TransactionalMessageCheckService，这个服务是周期检查事务的服务，
        this.transactionalMessageCheckService = new TransactionalMessageCheckService(this);
    }


        private void initialAcl() {
        if (!this.brokerConfig.isAclEnable()) {
            log.info("The broker dose not enable acl");
            return;
        }
        //初始化AccessValidator对应的实现PlainAccessValidator
        List<AccessValidator> accessValidators = ServiceProvider.load(ServiceProvider.ACL_VALIDATOR_ID, AccessValidator.class);
        if (accessValidators == null || accessValidators.isEmpty()) {
            log.info("The broker dose not load the AccessValidator");
            return;
        }
        //把对应的权限校验加入对应的校验器中
        for (AccessValidator accessValidator: accessValidators) {
            final AccessValidator validator = accessValidator;
            accessValidatorMap.put(validator.getClass(),validator);
            this.registerServerRPCHook(new RPCHook() {

                @Override
                public void doBeforeRequest(String remoteAddr, RemotingCommand request) {
                    //Do not catch the exception
                    validator.validate(validator.parse(request, remoteAddr));
                }
                
                @Override
                public void doAfterResponse(String remoteAddr, RemotingCommand request, RemotingCommand response) {
                }
            });
        }
    }

     private void initialRpcHooks() {
        //加载对应的RPC钩子方法
        List<RPCHook> rpcHooks = ServiceProvider.load(ServiceProvider.RPC_HOOK_ID, RPCHook.class);
        if (rpcHooks == null || rpcHooks.isEmpty()) {
            return;
        }
        for (RPCHook rpcHook: rpcHooks) {
            //注册钩子方法
            this.registerServerRPCHook(rpcHook);
        }
    }

initialize方法中的逻辑比较多，主要可以分为以下几步：

服务器内的相关日志文件的加载，{user.home}/store/config/ 文件目录下的json配置文件（包含topics，consumerOffset，subscriptionGroup，consumerFilter）。以及通过DefaultMessageStore来加载CommitLog，ConsumeQueue，IndexFile等文件（这里关于DefaultMessageStore中对于CommitLog，ConsumeQueue，IndexFile等文件加载过程不进行讲解，可以看前面关于DefaultMessageStore的文章）
如果上述文件加载成功，会启动对应的Broker客户端，然后创建一些线程池，在后面注册 API 指令事件后会监听到API的时候会进行处理
注册事件到对应的Broker客户端上，然后会记录对应的API事件和对应线程池封装到一个对象中
启动一些定时任务，这些任务比如记录Broker状态，消费进度持久化等任务
初始化一些服务，比如事务相关（周期检查事务），消息权限校验初始化和Rpc调用钩子相关服务。对应的服务加载方式是Java的SPI方式。

BrokerStartup启动之前的初始化逻辑就是这些了，其中有一些需要看前面的文章，和源码进行分析。这里就不太过深入，后面的BrokerStartup的启动解析下篇文章解析

RocketMQ源码解析——Broker部分之Broker启动过

从启动脚本到启动类

启动类`BrokerStartup`

启动的准备工作`createBrokerController`

Broker端的核心`BrokerController`

`BrokerController`的构造方法

`BrokerController`的初始化`initialize`

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

RocketMQ源码解析——Broker部分之Broker启动过

从启动脚本到启动类

启动类BrokerStartup

启动的准备工作createBrokerController

Broker端的核心BrokerController

BrokerController的构造方法

BrokerController的初始化initialize

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

启动类`BrokerStartup`

启动的准备工作`createBrokerController`

Broker端的核心`BrokerController`

`BrokerController`的构造方法

`BrokerController`的初始化`initialize`