RocketMQ源码9- broker 启动流程

墨蓝 2023-09-25 18:36 27阅读 0赞

## 1. 启动入口 ##

`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.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;
    }
    复制代码

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

1.  处理配置：主要是处理`nettyServerConfig`与`nettyClientConfig`的配置，这块就是一些配置解析的操作，处理方式与`NameServer`很类似，这里就不多说了。
2.  创建及初始化`controller`：调用方法`controller.initialize()`，这块内容我们后面分析。
3.  注册关闭钩子：调用`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#initi`