[RocketMQ] Broker启动流程源码解析 (二)

文章目录

- - 1.Brocker介绍
  - 2.BrokerStartup启动入口
  - 3.createBrokerController创建BrokerController
  - - 3.1 创建各种配置类
    - 3.2 创建broker控制器
    - 3.3 初始化broker控制器
    - - 3.3.1 加载配置文件
      - 3.3.2 创建消息存储对象MessageStore
      - 3.3.3 Load加载恢复消息文件
      - 3.3.4 初始化Broker通信层
      - 3.3.5 创建各种执行器线程池
      - 3.3.6 注册netty消息处理器
      - 3.3.7 启动定时周期任务
      - 3.3.8 初始化事务消息相关服务
      - 3.3.9 初始化ACL权限服务
      - 3.3.10 初始化RpCHook
    - 3.4 多端口监听
  - 4.Start启动BrokerController
  - 5.Broker启动流程总结

1.Brocker介绍

在这里插入图片描述

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

Remoting Module: 整个Broker的实体, 负责处理来自Client端的请求
Client Manager: 负责管理客户端 Producer和Consumer, 维护Consumer的Topic订阅信息
Store Service: 提供方便简单的API接口处理消息存储到物理硬盘和查询功能。
HA Service: 高可用服务, 提供Master Broker和Slave Broker的数据同步功能。
Index Service: 根据特定的Message key对投递到Broker的消息进行索引服务, 从而提供快速的消息查询。

2.BrokerStartup启动入口

该方法将会创建并且初始化一个BrokerController实例

在这里插入图片描述

3.createBrokerController创建BrokerController

createBrokerController创建BrokerController: 解析命令行, 加载Broker配置, NettyServer配置, NettyClient配置, 并保持, 然后进行一些配置的校验, 日志的配置, 然后创建一个BrokerController实例。

在这里插入图片描述

BrokerController相当于Broker的一个中央控制器类, 创建了BrokerController实例后, 执行initialize方法进行初始化操作。

public static BrokerController createBrokerController(String[] args) {
    //设置RocketMQ的版本信息，设置属性rocketmq.remoting.version，即当前rocketmq版本
    System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));

    try {
        //PackageConflictDetect.detectFastjson();
        /*
         * 1 jar包启动时，构建命令行操作的指令，使用main方法启动可以忽略
         */
        Options options = ServerUtil.buildCommandlineOptions(new Options());
        //mqbroker命令文件
        commandLine = ServerUtil.parseCmdLine("mqbroker", args, buildCommandlineOptions(options),
                new PosixParser());
        if (null == commandLine) {
            System.exit(-1);
        }
        /*
         * 2 创建broker的配置类
         */
        //创建Broker的配置类，包含Broker的各种配置，比如ROCKETMQ_HOME
        final BrokerConfig brokerConfig = new BrokerConfig();
        //NettyServer的配置类，Broker作为服务端，比如接收来自客户端的消息的时候
        final NettyServerConfig nettyServerConfig = new NettyServerConfig();
        //NettyClient的配置类，Broker还会作为客户端，比如连接NameServer的时候
        final NettyClientConfig nettyClientConfig = new NettyClientConfig();
        // tls安全相关配置
        nettyClientConfig.setUseTLS(Boolean.parseBoolean(System.getProperty(TLS_ENABLE,
                String.valueOf(TlsSystemConfig.tlsMode == TlsMode.ENFORCING))));
        //设置作为NettyServer时的监听端口为10911
        nettyServerConfig.setListenPort(10911);
        //Broker的消息存储配置，例如各种文件大小等
        final MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
        //如果broker的角色是slave，设置命中消息在内存的最大比例
        //默认broker角色是异步master
        if (BrokerRole.SLAVE == messageStoreConfig.getBrokerRole()) {
            //30
            int ratio = messageStoreConfig.getAccessMessageInMemoryMaxRatio() - 10;
            messageStoreConfig.setAccessMessageInMemoryMaxRatio(ratio);
        }
        /*
         * 3 解析外部配置文件
         * 判断命令行中是否包含字符'c'，即是否包含通过命令行指定配置文件的命令
         * 例如，启动Broker的时候添加的 -c /Volumes/Samsung/Idea/rocketmq/config/conf/broker.conf命令
         */
        if (commandLine.hasOption('c')) {
            //获取该命令指定的配置文件
            String file = commandLine.getOptionValue('c');
            if (file != null) {
                //加载外部配置文件
                configFile = file;
                InputStream in = new BufferedInputStream(new FileInputStream(file));
                properties = new Properties();
                properties.load(in);
                //将rmqAddressServerDomain、rmqAddressServerSubGroup属性设置为系统属性
                properties2SystemEnv(properties);
                //设置broker的配置信息
                MixAll.properties2Object(properties, brokerConfig);
                //设置nettyServer的配置信息
                MixAll.properties2Object(properties, nettyServerConfig);
                //设置nettyClient的配置信息
                MixAll.properties2Object(properties, nettyClientConfig);
                //设置messageStore的配置信息
                MixAll.properties2Object(properties, messageStoreConfig);
                //设置配置文件路径
                BrokerPathConfigHelper.setBrokerConfigPath(file);
                in.close();
            }
        }
        //设置broker的配置信息
        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);
        }
        /*
         * 4 获取namesrvAddr，即NameServer的地址，并进行校验
         */
        String namesrvAddr = brokerConfig.getNamesrvAddr();
        if (null != namesrvAddr) {
            try {
                //拆分NameServer的地址
                //可以指定多个NameServer的地址，以";"分隔
                String[] addrArray = namesrvAddr.split(";");
                //将字符串的地址，转换为网络连接的SocketAddress，检测格式是否正确
                for (String addr : addrArray) {
                    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);
            }
        }
        /*
         * 4 设置、校验brokerId
         *
         * 根据broker的角色配置brokerId，默认角色是ASYNC_MASTER
         * 通过此配置可知BrokerId为0表示Master，非0表示Slave
         */
        switch (messageStoreConfig.getBrokerRole()) {
            case ASYNC_MASTER:
            case SYNC_MASTER:
                //如果是master角色，那么设置brokerId为0
                brokerConfig.setBrokerId(MixAll.MASTER_ID);
                break;
            case SLAVE:
                //如果是slave角色，需要brokerId大于0
                if (brokerConfig.getBrokerId() <= 0) {
                    System.out.printf("Slave's brokerId must be > 0");
                    System.exit(-3);
                }

                break;
            default:
                break;
        }
        // 开启 DLeger 的操作
        if (messageStoreConfig.isEnableDLegerCommitLog()) {
            brokerConfig.setBrokerId(-1);
        }
        /*
         * 5 设置高可用通信监听端口，为监听端口+1，默认就是10912
         * 该端口主要用于比如主从同步之类的高可用操作
         *
         * 在配置broker集群的时候需要注意，配置集群时可能会抛出：Address already in use
         * 因为一个broker机器会占用三个端口，监听ip端口，以及监听ip端口+1的端口，监听ip端口-2端口
         */
        messageStoreConfig.setHaListenPort(nettyServerConfig.getListenPort() + 1);
        /*
         * 6 日志相关配置
         */
        LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
        //Joran 是 logback 使用的一个配置加载库，可以直接调用JoranConfigurator类重新实现logback的配置机制，
        JoranConfigurator configurator = new JoranConfigurator();
        configurator.setContext(lc);
        lc.reset();
        //配置broker日志文件的路徑
        System.setProperty("brokerLogDir", "");
        //isolateLogEnable属性表示在同一台机器上部署多个broker时是否区分日志路径，默認false
        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");

        /*判断命令行中是否包含字符'p'（printConfigItem）和'm'，如果存在则打印配置信息并结束jvm运行，没有的话就不用管*/
        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')) {
            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);
        }

        //打印当前broker的配置日志
        log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
        MixAll.printObjectProperties(log, brokerConfig);
        MixAll.printObjectProperties(log, nettyServerConfig);
        MixAll.printObjectProperties(log, nettyClientConfig);
        MixAll.printObjectProperties(log, messageStoreConfig);

        /*
         * 7 实例化BrokerController，设置各种属性
         */
        final BrokerController controller = new BrokerController(
                brokerConfig,
                nettyServerConfig,
                nettyClientConfig,
                messageStoreConfig);
        // 将所有的-c的外部配置信息保存到NamesrvController中的Configuration对象属性的allConfigs属性中
        controller.getConfiguration().registerConfig(properties);
        /*
         * 8 初始化BrokerController
         * 创建netty远程服务，初始化Netty线程池，注册请求处理器，配置定时任务，用于扫描并移除不活跃的Broker等操作。
         */
        boolean initResult = controller.initialize();
        //初始化失败则退出
        if (!initResult) {
            controller.shutdown();
            System.exit(-3);
        }
        /*
         * 9 添加关闭钩子方法，在Broker关闭之前执行，进行一些内存清理、对象销毁等操作
         */
        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();
                        //执行controller的shutdown方法，并且还会在messageStore#shutdown方法中将abort临时文件删除。
                        controller.shutdown();
                        long consumingTimeTotal = System.currentTimeMillis() - beginTime;
                        log.info("Shutdown hook over, consuming total time(ms): {}", consumingTimeTotal);
                    }
                }
            }
        }, "ShutdownHook"));
        //返回BrokerController
        return controller;
    } catch (Throwable e) {
        e.printStackTrace();
        System.exit(-1);
    }

    return null;
}

3.1 创建各种配置类

BrokerConfig: Broker的配置类, ROCKETMQ_HOME, namesrvAddr, brokerName, brokerId等属性。
NettyServerConfig: NettyServer的配置类, 包含Broker作为服务端时的各种属性, 比如客户端进行交互的时候, NettyServer的端口号为10911, 客户端和Broker通信时使用的是10911端口。
NettyClientConfig: NettyClient的配置类, 包含Broker作为客户端时的各种属性, Broker和NameServer交互的时候作为客户端。
MessageStoreConfig: Broker消息存储的配置类, 包含消息存储的相关配置, 比如文件的目录、大小等。

设置了配置信息之后, 会进行一系列校验:

在这里插入图片描述

ROCKETMQ_HOME校验: 如果启动参数中没有指定ROCKETMQ_HOME属性, 那么打印异常并退出, ROCKETMQ_HOME就是指定RocketMQ的配置文件路径。
namesrvAddr校验: Broker会通过将各个Name server 的字符串地址转换为InetSocketAddress来校验各个地址的合法性。
brokerId校验: 如果broker是同步master或者异步master, 设置brokerId为0, 如果是slave, 且brokerId不大于0, 则打印异常, 退出程序。

3.2 创建broker控制器

BrokerController的构造器实际上同样是在进行一系列的赋值和初始化操作, 创建各种manager和queue组件, 组件角色之间的互动是通过BrokerController完成的, 不是组件直接完成的。

在这里插入图片描述

public BrokerController(
    final BrokerConfig brokerConfig,
    final NettyServerConfig nettyServerConfig,
    final NettyClientConfig nettyClientConfig,
    final MessageStoreConfig messageStoreConfig
) {
    //broker的配置
    this.brokerConfig = brokerConfig;
    //作为netty服务端与客户端交互的配置
    this.nettyServerConfig = nettyServerConfig;
    //作为netty客户端与服务端交互的配置
    this.nettyClientConfig = nettyClientConfig;
    //消息存储的配置
    this.messageStoreConfig = messageStoreConfig;
    //消费者偏移量管理器，维护offset进度信息
    this.consumerOffsetManager = messageStoreConfig.isEnableLmq() ? new LmqConsumerOffsetManager(this) : new ConsumerOffsetManager(this);
    //topic配置管理器，管理broker中存储的所有topic的配置
    this.topicConfigManager = messageStoreConfig.isEnableLmq() ? new LmqTopicConfigManager(this) : new TopicConfigManager(this);
    //拉取消息处理器，用于处理拉取消息的请求
    this.pullMessageProcessor = new PullMessageProcessor(this);
    //拉取请求挂起服务，处理无消息时push长轮询消费者的挂起等待机制
    this.pullRequestHoldService = messageStoreConfig.isEnableLmq() ? new LmqPullRequestHoldService(this) : new PullRequestHoldService(this);
    //消息送达的监听器，生产者消息到达时通过该监听器触发pullRequestHoldService通知pullRequestHoldService
    this.messageArrivingListener = new NotifyMessageArrivingListener(this.pullRequestHoldService);
    //消费者id变化监听器
    this.consumerIdsChangeListener = new DefaultConsumerIdsChangeListener(this);
    //消费者管理类，维护消费者组的注册实例信息以及topic的订阅信息，并对消费者id变化进行监听
    this.consumerManager = new ConsumerManager(this.consumerIdsChangeListener);
    //消费者过滤管理器，配置文件为：xx/config/consumerFilter.json
    this.consumerFilterManager = new ConsumerFilterManager(this);
    //生产者管理器，包含生产者的注册信息，通过groupName分组
    this.producerManager = new ProducerManager();
    //客户端连接心跳服务，用于定时扫描生产者和消费者客户端，并将不活跃的客户端通道及相关信息移除
    this.clientHousekeepingService = new ClientHousekeepingService(this);
    //处理某些broker到客户端的请求，例如检查生产者的事务状态，重置offset
    this.broker2Client = new Broker2Client(this);
    //订阅分组关系管理器，维护消费者组的一些附加运维信息
    this.subscriptionGroupManager = messageStoreConfig.isEnableLmq() ? new LmqSubscriptionGroupManager(this) : new SubscriptionGroupManager(this);
    //broker对方访问的API，处理broker对外的发起请求，比如向nameServer注册，向master、slave发起的请求
    this.brokerOuterAPI = new BrokerOuterAPI(nettyClientConfig);
    //过滤服务管理器，拉取消息过滤
    this.filterServerManager = new FilterServerManager(this);
    //用于从节点，定时向主节点发起请求同步数据，例如topic配置、消费位移等
    this.slaveSynchronize = new SlaveSynchronize(this);

    /*初始化各种阻塞队列。将会被设置到对应的处理不同客户端请求的线程池执行器中*/
    //处理来自生产者的发送消息的请求的队列
    this.sendThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getSendThreadPoolQueueCapacity());
    //https://github.com/apache/rocketmq/pull/3631
    this.putThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getPutThreadPoolQueueCapacity());
    //处理来自消费者的拉取消息的请求的队列
    this.pullThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getPullThreadPoolQueueCapacity());
    //处理reply消息的请求的队列，RocketMQ4.7.0版本中增加了request-reply新特性，该特性允许producer在发送消息后同步或者异步等待consumer消费完消息并返回响应消息，类似rpc调用效果。
    //即生产者发送了消息之后，可以同步或者异步的收到消费了这条消息的消费者的响应
    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());
    //broker状态管理器，保存Broker运行时状态
    this.brokerStatsManager = messageStoreConfig.isEnableLmq() ? new LmqBrokerStatsManager(this.brokerConfig.getBrokerClusterName(), this.brokerConfig.isEnableDetailStat()) : new BrokerStatsManager(this.brokerConfig.getBrokerClusterName(), this.brokerConfig.isEnableDetailStat());
    //目前没用到
    this.setStoreHost(new InetSocketAddress(this.getBrokerConfig().getBrokerIP1(), this.getNettyServerConfig().getListenPort()));
    //broker快速失败服务
    this.brokerFastFailure = new BrokerFastFailure(this);
    //配置类
    this.configuration = new Configuration(
        log,
        BrokerPathConfigHelper.getBrokerConfigPath(),
        this.brokerConfig, this.nettyServerConfig, this.nettyClientConfig, this.messageStoreConfig
    );
}

3.3 初始化broker控制器

创建broker控制器以后, 需要进行broker的初始化

加载 config 路径下的json配置文件, 包括topic主题配置、offset消费偏移量、subscriptionGroup订阅分组、consumerFilter消费者过滤等配置文件。
加载实例化消息存储相关类DefaultMessageStore, 通过该类的load方法加载消息存储的相关文件, ，比如commitLog日志文件、consumequeue消息消费队列文件, indexFile索引文件, messageStore还会将这些文件的内容加载到内存中, 完成RocketMq的数据恢复。

大致过程:

加载配置文件: topic配置文件, offset消费偏移量、subscriptionGroup订阅分组、consumerFilter消费者过滤等配置文件。
实例化和初始化消息存储服务相关类DefaultMessageStore。
该类的load方法加载消息存储的相关文件, ，比如commitLog日志文件、consumequeue消息消费队列文件, indexFile索引文件, messageStore还会将这些文件的内容加载到内存中, 完成RocketMq的数据恢复。
创建netty远程服务, 包括remotingServer和fastRemotingServer。
调用registerProcessor方法注册netty消息处理器到netty远程服务中。
创建一系列定时任务, 用于检查broker状态, 统计接收, 拉取消息数量, 持久化消息文件, 修改na’meServer地址。
调用initialTransaction初始化事务消息相关服务。
调用initialAcl初始化权限相关服务。
调用initialRpcHooks初始化所有RpcHook钩子。

public boolean initialize() throws CloneNotSupportedException {
   /*
    * 1 加载配置文件
    * 尝试从json配置文件或者bak备份文件中加载json字符串，然后反序列化转换为自身内部的属性
    */
   //topic配置文件加载，路径为  {user.home}/store/config/topics.json
   boolean result = this.topicConfigManager.load();
   //消费者消费偏移量配置文件加载，路径为  {user.home}/store/config/consumerOffset.json
   result = result && this.consumerOffsetManager.load();
   //订阅分组配置文件加载，路径为  {user.home}/store/config/subscriptionGroup.json
   result = result && this.subscriptionGroupManager.load();
   //消费者过滤配置文件加载，路径为  {user.home}/store/config/consumerFilter.json
   result = result && this.consumerFilterManager.load();

   /*
    * 如果上一步加载配置全部成功
    * 2 实例化和初始化消息存储服务相关类DefaultMessageStore
    */
   if (result) {
       try {
           //实例化消息存储类DefaultMessageStore
           this.messageStore =
                   new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener,
                           this.brokerConfig);
           //如果启动了enableDLegerCommitLog，就创建DLeger组件DLedgerRoleChangeHandler。默认为false，如果需要开启则该值需要设置为true。
           //在启用enableDLegerCommitLog情况下，broker通过raft协议选主，可以实现主从角色自动切换，这是4.5版本之后的新功能
           //简单的说，如果启动enableDLegerCommitLog，就表示启用 RocketMQ 的容灾机制——自动主从切换，
           if (messageStoreConfig.isEnableDLegerCommitLog()) {
               DLedgerRoleChangeHandler roleChangeHandler = new DLedgerRoleChangeHandler(this, (DefaultMessageStore) messageStore);
               ((DLedgerCommitLog) ((DefaultMessageStore) messageStore).getCommitLog()).getdLedgerServer().getdLedgerLeaderElector().addRoleChangeHandler(roleChangeHandler);
           }
           //broker的统计服务类，保存了broker的一些统计数据
           //例如msgPutTotalTodayNow --现在存储的消息数量，msgPutTotalTodayMorning --今天存储的消息数量
           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);
       }
   }
   /*
    * 如果上一步加载配置全部成功
    * 3 通过消息存储服务加载消息存储的相关文件，比如commitLog日志文件、consumequeue消息消费队列文件的加载，indexFile索引文件的构建
    * messageStore还会将这些文件的内容加载到内存中，并且完成RocketMQ的数据恢复
    * 这是核broker启动的心步骤之一
    */
   result = result && this.messageStore.load();

   /*
    * 如果上一步加载配置全部成功
    * 3 开始初始化Broker通信层和各种请求执行器
    */
   if (result) {
       /*
        * 4 创建netty远程服务，remotingServer和fastRemotingServer
        */
       //创建broker的netty远程服务，端口为10911，可以用于处理客户端的所有请求。
       this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.clientHousekeepingService);
       //创建一个broker的netty快速远程服务的配置
       NettyServerConfig fastConfig = (NettyServerConfig) this.nettyServerConfig.clone();
       //设置快速netty远程服务的配置监听的端口号为普通服务的端口-2，默认10909
       fastConfig.setListenPort(nettyServerConfig.getListenPort() - 2);
       //创建broker的netty快速服务，这就是所谓的快速通道，对应可以处理客户端除了拉取消息之外的所有请求，所谓的VIP端口。
       this.fastRemotingServer = new NettyRemotingServer(fastConfig, this.clientHousekeepingService);
       /*
        * 5 创建各种执行器线程池
        */
       //处理发送消息的请求的线程池
       this.sendMessageExecutor = new BrokerFixedThreadPoolExecutor(
               this.brokerConfig.getSendMessageThreadPoolNums(),
               this.brokerConfig.getSendMessageThreadPoolNums(),
               1000 * 60,
               TimeUnit.MILLISECONDS,
               this.sendThreadPoolQueue,
               new ThreadFactoryImpl("SendMessageThread_"));
       ///https://github.com/apache/rocketmq/pull/3631
       this.putMessageFutureExecutor = new BrokerFixedThreadPoolExecutor(
               this.brokerConfig.getPutMessageFutureThreadPoolNums(),
               this.brokerConfig.getPutMessageFutureThreadPoolNums(),
               1000 * 60,
               TimeUnit.MILLISECONDS,
               this.putThreadPoolQueue,
               new ThreadFactoryImpl("PutMessageThread_"));
       //处理拉取消息的请求的线程池
       this.pullMessageExecutor = new BrokerFixedThreadPoolExecutor(
               this.brokerConfig.getPullMessageThreadPoolNums(),
               this.brokerConfig.getPullMessageThreadPoolNums(),
               1000 * 60,
               TimeUnit.MILLISECONDS,
               this.pullThreadPoolQueue,
               new ThreadFactoryImpl("PullMessageThread_"));
       //处理reply消息的请求的线程池
       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_"));
       //broker 管理线程池，作为默认处理器的线程池
       this.adminBrokerExecutor =
               Executors.newFixedThreadPool(this.brokerConfig.getAdminBrokerThreadPoolNums(), new ThreadFactoryImpl(
                       "AdminBrokerThread_"));
       //客户端管理器的线程池
       this.clientManageExecutor = new ThreadPoolExecutor(
               this.brokerConfig.getClientManageThreadPoolNums(),
               this.brokerConfig.getClientManageThreadPoolNums(),
               1000 * 60,
               TimeUnit.MILLISECONDS,
               this.clientManagerThreadPoolQueue,
               new ThreadFactoryImpl("ClientManageThread_"));
       //心跳处理的线程池
       this.heartbeatExecutor = new BrokerFixedThreadPoolExecutor(
               this.brokerConfig.getHeartbeatThreadPoolNums(),
               this.brokerConfig.getHeartbeatThreadPoolNums(),
               1000 * 60,
               TimeUnit.MILLISECONDS,
               this.heartbeatThreadPoolQueue,
               new ThreadFactoryImpl("HeartbeatThread_", true));
       //事务消息相关处理的线程池
       this.endTransactionExecutor = new BrokerFixedThreadPoolExecutor(
               this.brokerConfig.getEndTransactionThreadPoolNums(),
               this.brokerConfig.getEndTransactionThreadPoolNums(),
               1000 * 60,
               TimeUnit.MILLISECONDS,
               this.endTransactionThreadPoolQueue,
               new ThreadFactoryImpl("EndTransactionThread_"));
       //消费者管理的线程池
       this.consumerManageExecutor =
               Executors.newFixedThreadPool(this.brokerConfig.getConsumerManageThreadPoolNums(), new ThreadFactoryImpl(
                       "ConsumerManageThread_"));

       /*
        * 6 注册处理器
        */
       this.registerProcessor();
       /*
        * 7 启动一系列定时周期任务
        */
       final long initialDelay = UtilAll.computeNextMorningTimeMillis() - System.currentTimeMillis();
       //每隔24h打印昨天生产和消费的消息数量
       final long period = 1000 * 60 * 60 * 24;
       this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
           @Override
           public void run() {
               try {
                   BrokerController.this.getBrokerStats().record();
               } catch (Throwable e) {
                   log.error("schedule record error.", e);
               }
           }
       }, initialDelay, period, TimeUnit.MILLISECONDS);
       //每隔5s將消费者offset进行持久化，存入consumerOffset.json文件中
       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);
       //每隔10s將消费过滤信息进行持久化，存入consumerFilter.json文件中
       this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
           @Override
           public void run() {
               try {
                   BrokerController.this.consumerFilterManager.persist();
               } catch (Throwable e) {
                   log.error("schedule persist consumer filter error.", e);
               }
           }
       }, 1000 * 10, 1000 * 10, TimeUnit.MILLISECONDS);
       //每隔3m將检查消费者的消费进度
       //当消费进度落后阈值的时候，并且disableConsumeIfConsumerReadSlowly=true(默认false)，就停止消费者消费，保护broker，避免消费积压
       this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
           @Override
           public void run() {
               try {
                   BrokerController.this.protectBroker();
               } catch (Throwable e) {
                   log.error("protectBroker error.", e);
               }
           }
       }, 3, 3, TimeUnit.MINUTES);
       //每隔1s將打印发送消息线程池队列、拉取消息线程池队列、查询消息线程池队列、结束事务线程池队列的大小以及队列头部元素存在时间
       this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
           @Override
           public void run() {
               try {
                   BrokerController.this.printWaterMark();
               } catch (Throwable e) {
                   log.error("printWaterMark error.", e);
               }
           }
       }, 10, 1, TimeUnit.SECONDS);
       //每隔1m將打印已存储在commitlog提交日志中但尚未分派到consume queue消费队列的字节数。
       this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

           @Override
           public void run() {
               try {
                   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);

       //如果broker的nameServer地址不为null
       if (this.brokerConfig.getNamesrvAddr() != null) {
           //更新nameServer地址
           this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr());
           log.info("Set user specified name server address: {}", this.brokerConfig.getNamesrvAddr());
       } else if (this.brokerConfig.isFetchNamesrvAddrByAddressServer()) {
           //如果没有指定nameServer地址，并且允许从地址服务器获取nameServer地址
           //那么每隔2m从nameServer地址服务器拉取最新的nameServer地址并更新
           //要想动态更新nameServer地址，需要指定一个地址服务器的url，并且fetchNamesrvAddrByAddressServer设置为true
           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服务，DLeger开启后表示支持高可用的主从自动切换
       if (!messageStoreConfig.isEnableDLegerCommitLog()) {
           //如果当前broker是slave从节点
           if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
               //根据是否配置了HA地址，来更新HA地址，并设置updateMasterHAServerAddrPeriodically
               if (this.messageStoreConfig.getHaMasterAddress() != null && this.messageStoreConfig.getHaMasterAddress().length() >= 6) {
                   this.messageStore.updateHaMasterAddress(this.messageStoreConfig.getHaMasterAddress());
                   this.updateMasterHAServerAddrPeriodically = false;
               } else {
                   this.updateMasterHAServerAddrPeriodically = true;
               }
           } else {
               //如果是主节点，每隔60s將打印主从节点的差异
               this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
                   @Override
                   public void run() {
                       try {
                           BrokerController.this.printMasterAndSlaveDiff();
                       } catch (Throwable e) {
                           log.error("schedule printMasterAndSlaveDiff error.", e);
                       }
                   }
               }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
           }
       }
       //Tls传输相关配置，通信安全的文件监听模块，用来观察网络加密配置文件的更改
       //默认是PERMISSIVE，因此会进入代码块
       if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {
           // Register a listener to reload SslContext
           try {
               //实例化文件监听服务,并且初始化事务消息服务。
               fileWatchService = new FileWatchService(
                       new String[]{
                               TlsSystemConfig.tlsServerCertPath,
                               TlsSystemConfig.tlsServerKeyPath,
                               TlsSystemConfig.tlsServerTrustCertPath
                       },
                       new FileWatchService.Listener() {
                           boolean certChanged, keyChanged = false;

                           @Override
                           public 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");
           }
       }
       /*
        * 8 初始化事务消息相关服务
        */
       initialTransaction();
       /*
        * 9 初始化权限相关服务
        */
       initialAcl();
       /*
        * 10 初始化RPC调用的钩子函数
        */
       initialRpcHooks();
   }
   return result;
}

3.3.1 加载配置文件

load方法加载配置文件。加载 config 路径下的json配置文件, 包括topic主题配置、offset消费偏移量、subscriptionGroup订阅分组、consumerFilter消费者过滤等配置文件。

在这里插入图片描述

3.3.2 创建消息存储对象MessageStore

DefaultMessageStore是RocketMQ的核心文件存储控制类, 是RocketMQ对于消息存储和获取功能的抽象。DefaultMessageStore类位于store模块, 可以直接控制管理commitLog, consumerQueue, indexFile文件的读和写。

启动Broker时候, 会创建一个DefaultMessageStore对象, load方法进行磁盘文件的加载和异常数据的修复。

public DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager 
brokerStatsManager,final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig) throws IOException {
      //消息送达的监听器，生产者消息到达时通过该监听器触发pullRequestHoldService通知pullRequestHoldService
      this.messageArrivingListener = messageArrivingListener;
      //Broker的配置类，包含Broker的各种配置，比如ROCKETMQ_HOME
      this.brokerConfig = brokerConfig;
      //Broker的消息存储配置，例如各种文件大小等
      this.messageStoreConfig = messageStoreConfig;
      //broker状态管理器，保存Broker运行时状态，统计工作
      this.brokerStatsManager = brokerStatsManager;
      //创建 MappedFile文件的服务，用于初始化MappedFile和预热MappedFile
      this.allocateMappedFileService = new AllocateMappedFileService(this);
      //实例化CommitLog，DLedgerCommitLog表示主持主从自动切换功能，默认是CommitLog类型
      if (messageStoreConfig.isEnableDLegerCommitLog()) {
          this.commitLog = new DLedgerCommitLog(this);
      } else {
          this.commitLog = new CommitLog(this);
      }
      //topic的ConsumeQueueMap的对应关系
      this.consumeQueueTable = new ConcurrentHashMap<>(32);
      //ConsumeQueue文件的刷盘服务
      this.flushConsumeQueueService = new FlushConsumeQueueService();
      //清除过期CommitLog文件的服务
      this.cleanCommitLogService = new CleanCommitLogService();
      //清除过期ConsumeQueue文件的服务
      this.cleanConsumeQueueService = new CleanConsumeQueueService();
      //存储一些统计指标信息的服务
      this.storeStatsService = new StoreStatsService();
      //IndexFile索引文件服务
      this.indexService = new IndexService(this);
      //高可用服务，默认为null
      if (!messageStoreConfig.isEnableDLegerCommitLog()) {
          this.haService = new HAService(this);
      } else {
          this.haService = null;
      }
      //根据CommitLog文件，更新index文件索引和ConsumeQueue文件偏移量的服务
      this.reputMessageService = new ReputMessageService();
      //处理RocketMQ延迟消息的服务
      this.scheduleMessageService = new ScheduleMessageService(this);
      //初始化MappedFile的时候进行ByteBuffer的分配回收
      this.transientStorePool = new TransientStorePool(messageStoreConfig);
      //如果当前节点不是从节点，并且是异步刷盘策略，并且transientStorePoolEnable参数配置为true，则启动该服务
      if (messageStoreConfig.isTransientStorePoolEnable()) {
          this.transientStorePool.init();
      }
      //启动MappedFile文件服务线程
      this.allocateMappedFileService.start();
      //启动index索引文件服务线程
      this.indexService.start();

      //转发服务列表，监听CommitLog文件中的新消息存储，然后会调用列表中的CommitLogDispatcher#dispatch方法
      this.dispatcherList = new LinkedList<>();
      //通知ConsumeQueue的Dispatcher，可用于更新ConsumeQueue的偏移量等信息
      this.dispatcherList.addLast(new CommitLogDispatcherBuildConsumeQueue());
      //通知IndexFile的Dispatcher，可用于更新IndexFile的时间戳等信息
      this.dispatcherList.addLast(new CommitLogDispatcherBuildIndex());
      //获取锁文件，路径就是配置的{storePathRootDir}/lock
      File file = new File(StorePathConfigHelper.getLockFile(messageStoreConfig.getStorePathRootDir()));
      //确保创建file文件的父目录，即{storePathRootDir}目录
      MappedFile.ensureDirOK(file.getParent());
      //确保创建commitlog目录，即{StorePathCommitLog}目录
      MappedFile.ensureDirOK(getStorePathPhysic());
      //确保创建consumequeue目录，即{storePathRootDir}/consumequeue目录
      MappedFile.ensureDirOK(getStorePathLogic());
      //创建lockfile文件，名为lock，权限是"读写"，这是一个锁文件，用于获取文件锁。
//文件锁用来保证磁盘上的这些存储文件同时只能有一个Broker的messageStore来操作。
      lockFile = new RandomAccessFile(file, "rw");
  }

lockFile: 锁文件消息。
allocateMappedFileService: 创建和预热MappedFile文件的服务。
scheduleMessageService: 消息文件刷盘或清理服务, 延迟消息处理服务。

3.3.3 Load加载恢复消息文件

加载实例化消息存储相关类DefaultMessageStore, 通过该类的load方法加载消息存储的相关文件, ，比如commitLog日志文件、consumequeue消息消费队列文件, indexFile索引文件, messageStore还会将这些文件的内容加载到内存中, 完成RocketMq的数据恢复。

在这里插入图片描述

调用isTempFileExist方法判断上次broker是否是正常退出, 如果是正常退出的话不会保留abort文件, 异常退出会保持。
加载CommitLog日志文件。CommitLog文件是真正存储消息内容的地方。
加载ConsumeQueue文件。ConsumeQueue文件可以看作是CommitLog的消息偏移量索引文件。
加载 index 索引文件。Index文件可以看作是CommitLog的消息时间范围索引文件。
恢复ConsumeQueue文件和CommitLog文件, 将正确的数据恢复到内存, 删除错误数据和文件。
加载RocketMQ延迟消息的服务, 包含延迟等级和配置文件等。

/**
 * DefaultMessageStore的方法
 * 加载Commit Log、Consume Queue、index file等文件，将数据加载到内存中，并完成数据的恢复
 *
 * @throws IOException
 */
public boolean load() {
    boolean result = true;

    try {
        /*
         * 1 判断上次broker是否是正常退出，如果是正常退出不会保留abort文件，异常退出则会
         *
         * Broker在启动时会创建{storePathRootDir}/abort文件，并且注册钩子函数：在JVM退出时删除abort文件。
         * 如果下一次启动时存在abort文件，说明Broker是异常退出的，文件数据可能不一直，需要进行数据修复。
         */
        boolean lastExitOK = !this.isTempFileExist();
        log.info("last shutdown {}", lastExitOK ? "normally" : "abnormally");

        /*
         * 2 加载Commit Log日志文件，目录路径取自broker.conf文件中的storePathCommitLog属性
         * Commit Log文件是真正存储消息内容的地方，单个文件默认大小1G。
         */
        // load Commit Log
        result = result && this.commitLog.load();
        /*
         * 2 加载Consume Queue文件，目录路径为{storePathRootDir}/consumequeue，文件组织方式为topic/queueId/fileName
         * Consume Queue文件可以看作是Commit Log的索引文件，其存储了它所属Topic的消息在Commit Log中的偏移量
         * 消费者拉取消息的时候，可以从Consume Queue中快速的根据偏移量定位消息在Commit Log中的位置。
         */
        // load Consume Queue
        result = result && this.loadConsumeQueue();

        if (result) {
            /*
             * 3 加载checkpoint 检查点文件，文件位置是{storePathRootDir}/checkpoint
             * StoreCheckpoint记录着commitLog、ConsumeQueue、Index文件的最后更新时间点，
             * 当上一次broker是异常结束时，会根据StoreCheckpoint的数据进行恢复，这决定着文件从哪里开始恢复，甚至是删除文件
             */
            this.storeCheckpoint =
                    new StoreCheckpoint(StorePathConfigHelper.getStoreCheckpoint(this.messageStoreConfig.getStorePathRootDir()));
            /*
             * 4 加载 index 索引文件，目录路径为{storePathRootDir}/index
             * index 索引文件用于通过时间区间来快速查询消息，底层为HashMap结构，实现为hash索引。后面会专门出文介绍
             * 如果不是正常退出，并且最大更新时间戳比checkpoint文件中的时间戳大，则删除该 index 文件
             */
            this.indexService.load(lastExitOK);
            /*
             * 4 恢复ConsumeQueue文件和CommitLog文件，将正确的的数据恢复至内存中，删除错误数据和文件。
             */
            this.recover(lastExitOK);

            log.info("load over, and the max phy offset = {}", this.getMaxPhyOffset());

            /*
             * 5 加载RocketMQ延迟消息的服务，包括延时等级、配置文件等等。
             */
            if (null != scheduleMessageService) {
                result = this.scheduleMessageService.load();
            }
        }

    } catch (Exception e) {
        log.error("load exception", e);
        result = false;
    }

    if (!result) {
        //如果上面的操作抛出异常，则文件服务停止
        this.allocateMappedFileService.shutdown();
    }

    return result;
}

3.3.4 初始化Broker通信层

配置文件和日志文件加载恢复完毕后, 初始化Broker通信层和各种请求执行器。

创建两个NettyRemotingServer, 一个是普通的远程服务remotingServer, 端口号为10911, 快速的远程服务fastRemotingServer, 端口号为10909。

//创建broker的netty远程服务，端口为10911，可以用于处理客户端的所有请求。
this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.clientHousekeepingService);
//创建一个broker的netty快速远程服务的配置
NettyServerConfig fastConfig = (NettyServerConfig) this.nettyServerConfig.clone();
//设置快速netty远程服务的配置监听的端口号为普通服务的端口-2，默认10909
fastConfig.setListenPort(nettyServerConfig.getListenPort() - 2);
//创建broker的netty快速服务，这就是所谓的快速通道，对应可以处理客户端除了拉取消息之外的所有请求，所谓的VIP端口。
this.fastRemotingServer = new NettyRemotingServer(fastConfig, this.clientHousekeepingService);

3.3.5 创建各种执行器线程池

因为RocketMQ为了性能, 将许多请求进行异步处理。

/*
 * 5 创建各种执行器线程池
 */
//处理发送消息的请求的线程池
this.sendMessageExecutor = new BrokerFixedThreadPoolExecutor(
        this.brokerConfig.getSendMessageThreadPoolNums(),
        this.brokerConfig.getSendMessageThreadPoolNums(),
        1000 * 60,
        TimeUnit.MILLISECONDS,
        this.sendThreadPoolQueue,
        new ThreadFactoryImpl("SendMessageThread_"));
///https://github.com/apache/rocketmq/pull/3631
this.putMessageFutureExecutor = new BrokerFixedThreadPoolExecutor(
        this.brokerConfig.getPutMessageFutureThreadPoolNums(),
        this.brokerConfig.getPutMessageFutureThreadPoolNums(),
        1000 * 60,
        TimeUnit.MILLISECONDS,
        this.putThreadPoolQueue,
        new ThreadFactoryImpl("PutMessageThread_"));
//处理拉取消息的请求的线程池
this.pullMessageExecutor = new BrokerFixedThreadPoolExecutor(
        this.brokerConfig.getPullMessageThreadPoolNums(),
        this.brokerConfig.getPullMessageThreadPoolNums(),
        1000 * 60,
        TimeUnit.MILLISECONDS,
        this.pullThreadPoolQueue,
        new ThreadFactoryImpl("PullMessageThread_"));
//处理reply消息的请求的线程池
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_"));
//broker 管理线程池，作为默认处理器的线程池
this.adminBrokerExecutor =
        Executors.newFixedThreadPool(this.brokerConfig.getAdminBrokerThreadPoolNums(), new ThreadFactoryImpl(
                "AdminBrokerThread_"));
//客户端管理器的线程池
this.clientManageExecutor = new ThreadPoolExecutor(
        this.brokerConfig.getClientManageThreadPoolNums(),
        this.brokerConfig.getClientManageThreadPoolNums(),
        1000 * 60,
        TimeUnit.MILLISECONDS,
        this.clientManagerThreadPoolQueue,
        new ThreadFactoryImpl("ClientManageThread_"));
//心跳处理的线程池
this.heartbeatExecutor = new BrokerFixedThreadPoolExecutor(
        this.brokerConfig.getHeartbeatThreadPoolNums(),
        this.brokerConfig.getHeartbeatThreadPoolNums(),
        1000 * 60,
        TimeUnit.MILLISECONDS,
        this.heartbeatThreadPoolQueue,
        new ThreadFactoryImpl("HeartbeatThread_", true));
//事务消息相关处理的线程池
this.endTransactionExecutor = new BrokerFixedThreadPoolExecutor(
        this.brokerConfig.getEndTransactionThreadPoolNums(),
        this.brokerConfig.getEndTransactionThreadPoolNums(),
        1000 * 60,
        TimeUnit.MILLISECONDS,
        this.endTransactionThreadPoolQueue,
        new ThreadFactoryImpl("EndTransactionThread_"));
//消费者管理的线程池
this.consumerManageExecutor =
        Executors.newFixedThreadPool(this.brokerConfig.getConsumerManageThreadPoolNums(), new ThreadFactoryImpl(
                "ConsumerManageThread_"));

3.3.6 注册netty消息处理器

registerProcessor方法中的所有的处理器都是通过remotingServer #registerProcessor方法进行注册的。

/**
 * BrokerController的方法
 * 注册netty消息处理器
 */
public void registerProcessor() {
    /**
     * 发送消息处理器
     */
    SendMessageProcessor sendProcessor = new SendMessageProcessor(this);
    sendProcessor.registerSendMessageHook(sendMessageHookList);
    sendProcessor.registerConsumeMessageHook(consumeMessageHookList);

    //对于发送类型的请求，使用发送消息处理器sendProcessor来处理
    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.fastRemotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor);
    this.fastRemotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor);
    this.fastRemotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendProcessor, this.sendMessageExecutor);
    this.fastRemotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor, this.sendMessageExecutor);
}

3.3.7 启动定时周期任务

注册了netty消息处理器之后, 会启动一系列的定时任务, 这些任务由BrokerController中的scheduledExecutorService执行, 该线程池只有一个线程。

private final ScheduledExecutorService scheduledExecutorService = 
Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl(
        "BrokerControllerScheduledThread"));

每隔24h打印昨天生产和消费的消息数量。
每隔5s將消费者offset进行持久化, 存入consumerOffset.json文件中。
每隔10s將消费过滤信息进行持久化, 存入consumerFilter.json文件中。
每隔3m將检查消费者的消费进度, 当消费进度落后阈值的时候, 并且disableConsumeIfConsumerReadSlowly=true(默认false), 那么停止消费组消费, 包含broker, 避免消息堆积。
每隔1s將打印发送消息线程池队列、拉取消息线程池队列、查询消息线程池队列、结束事务线程池队列的大小以及队列头部元素存在时间。
每隔1m將打印已存储在commitlog提交日志中但尚未分派到consume queue消费队列的字节数。
如果broker手动设置的nameServer地址为null, 且允许从地址服务器获取nameServer地址, 那么每隔2s从nameServer地址服务器拉取最新的nameServer并更新。
如果没有开启DLeger服务, 如果节点是主节点的话, 那么每隔60s打印主从节点的差异。

/*
 * 7 启动一系列定时周期任务
 */
final long initialDelay = UtilAll.computeNextMorningTimeMillis() - System.currentTimeMillis();
//每隔24h打印昨天生产和消费的消息数量
final long period = 1000 * 60 * 60 * 24;
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
        try {
            BrokerController.this.getBrokerStats().record();
        } catch (Throwable e) {
            log.error("schedule record error.", e);
        }
    }
}, initialDelay, period, TimeUnit.MILLISECONDS);
//每隔5s將消费者offset进行持久化，存入consumerOffset.json文件中
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);
//每隔10s將消费过滤信息进行持久化，存入consumerFilter.json文件中
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
        try {
            BrokerController.this.consumerFilterManager.persist();
        } catch (Throwable e) {
            log.error("schedule persist consumer filter error.", e);
        }
    }
}, 1000 * 10, 1000 * 10, TimeUnit.MILLISECONDS);
//每隔3m將检查消费者的消费进度
//当消费进度落后阈值的时候，并且disableConsumeIfConsumerReadSlowly=true(默认false)，就停止消费者消费，保护broker，避免消费积压
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
        try {
            BrokerController.this.protectBroker();
        } catch (Throwable e) {
            log.error("protectBroker error.", e);
        }
    }
}, 3, 3, TimeUnit.MINUTES);
//每隔1s將打印发送消息线程池队列、拉取消息线程池队列、查询消息线程池队列、结束事务线程池队列的大小以及队列头部元素存在时间
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
        try {
            BrokerController.this.printWaterMark();
        } catch (Throwable e) {
            log.error("printWaterMark error.", e);
        }
    }
}, 10, 1, TimeUnit.SECONDS);
//每隔1m將打印已存储在commitlog提交日志中但尚未分派到consume queue消费队列的字节数。
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

    @Override
    public void run() {
        try {
            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);

//如果broker的nameServer地址不为null
if (this.brokerConfig.getNamesrvAddr() != null) {
    //更新nameServer地址
    this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr());
    log.info("Set user specified name server address: {}", this.brokerConfig.getNamesrvAddr());
} else if (this.brokerConfig.isFetchNamesrvAddrByAddressServer()) {
    //如果没有指定nameServer地址，并且允许从地址服务器获取nameServer地址
    //那么每隔2m从nameServer地址服务器拉取最新的nameServer地址并更新
    //要想动态更新nameServer地址，需要指定一个地址服务器的url，并且fetchNamesrvAddrByAddressServer设置为true
    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服务，DLeger开启后表示支持高可用的主从自动切换
if (!messageStoreConfig.isEnableDLegerCommitLog()) {
    //如果当前broker是slave从节点
    if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
        //根据是否配置了HA地址，来更新HA地址，并设置updateMasterHAServerAddrPeriodically
        if (this.messageStoreConfig.getHaMasterAddress() != null && this.messageStoreConfig.getHaMasterAddress().length() >= 6) {
            this.messageStore.updateHaMasterAddress(this.messageStoreConfig.getHaMasterAddress());
            this.updateMasterHAServerAddrPeriodically = false;
        } else {
            this.updateMasterHAServerAddrPeriodically = true;
        }
    } else {
        //如果是主节点，每隔60s將打印主从节点的差异
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    BrokerController.this.printMasterAndSlaveDiff();
                } catch (Throwable e) {
                    log.error("schedule printMasterAndSlaveDiff error.", e);
                }
            }
        }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
    }
}

3.3.8 初始化事务消息相关服务

initialTransaction为初始化事务的方法, 事务消息的服务采用SPI的方式进行加载。

transactionalMessageService: 事务消息服务, 用于处理、检查事务消息。
transactionalMessageCheckListener: 事务消息检查监听器。监听回查消息。
transactionalMessageCheckService: 事务消息检查服务, 提供了事务消息回查的逻辑, 默认情况下为6s如果没有commit/rollback的事务消息才会触发事务回查, 而如果回查数超过15次则丢弃事务。

/**
 * BrokerController的方法
 * 初始化事务消息相关服务
 */
private void initialTransaction() {
    //基于Java的SPI机制，查找"META-INF/service/org.apache.rocketmq.broker.transaction.TransactionalMessageService"文件里面的SPI实现
    //事务消息服务
    this.transactionalMessageService = ServiceProvider.loadClass(ServiceProvider.TRANSACTION_SERVICE_ID, TransactionalMessageService.class);
    if (null == this.transactionalMessageService) {
        //如果沒有通过SPI指定具体的实现，那么使用默认实现，TransactionalMessageServiceImpl
        this.transactionalMessageService = new TransactionalMessageServiceImpl(new TransactionalMessageBridge(this, this.getMessageStore()));
        log.warn("Load default transaction message hook service: {}", TransactionalMessageServiceImpl.class.getSimpleName());
    }
    //基于Java的SPI机制，查找"META-INF/service/org.apache.rocketmq.broker.transaction.AbstractTransactionalMessageCheckListener"文件里面的SPI实现
    //事务消息回查服务监听器
    this.transactionalMessageCheckListener = ServiceProvider.loadClass(ServiceProvider.TRANSACTION_LISTENER_ID, AbstractTransactionalMessageCheckListener.class);
    if (null == this.transactionalMessageCheckListener) {
        //如果沒有通过SPI指定具体的实现，那么使用默认实现，DefaultTransactionalMessageCheckListener
        this.transactionalMessageCheckListener = new DefaultTransactionalMessageCheckListener();
        log.warn("Load default discard message hook service: {}", DefaultTransactionalMessageCheckListener.class.getSimpleName());
    }
    this.transactionalMessageCheckListener.setBrokerController(this);
    //创建TransactionalMessageCheckService服务，该服务内部有一个线程
    //会定时每一分钟(可通过在broker.conf文件设置transactionCheckInterval属性更改)触发事务检查的逻辑，内部调用TransactionalMessageService#check方法
    //默认情况下，6秒以上没commit/rollback的事务消息才会触发事务回查，而如果回查次数超过15次则丢弃事务
    this.transactionalMessageCheckService = new TransactionalMessageCheckService(this);
}

3.3.9 初始化ACL权限服务

加载权限相关校验器, 注册进RpcHock, 在请求执行之前执行权限校验。

/**
 * BrokerController的方法
 * 初始化ACL权限校验器
 */
private void initialAcl() {
    //校验是否开启了ACL，默认false，所以直接返回了
    if (!this.brokerConfig.isAclEnable()) {
        log.info("The broker dose not enable acl");
        return;
    }
    //如果开启了ACL，则首先通过SPI机制获取AccessValidator
    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;
    }
    //将校验器存入accessValidatorMap，并且注册到RpcHook中，在请求之前会执行校验。
    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) {
            }
        });
    }
}

3.3.10 初始化RpCHook

基于Java的SPI机制查找RpCHook, 并注册到netty远程服务中。RpcHook是RocketMQ提供的钩子类, 可以在请求被处理之前和响应被返回之前执行对应的方法。

/**
 * BrokerController的方法
 * 初始化RpcHook服务
 */
private void initialRpcHooks() {
    //通过SPI机制获取RPCHook的实现
    List<RPCHook> rpcHooks = ServiceProvider.load(ServiceProvider.RPC_HOOK_ID, RPCHook.class);
    //如果没有配置RpcHook，那么直接返回
    if (rpcHooks == null || rpcHooks.isEmpty()) {
        return;
    }
    //遍历并且注册所有的RpcHook
    for (RPCHook rpcHook : rpcHooks) {
        this.registerServerRPCHook(rpcHook);
    }
}

// RpcHook是RocketMQ提供的钩子类
public interface RPCHook {
    /**
     * broker处理请求之前执行
     */
    void doBeforeRequest(final String remoteAddr, final RemotingCommand request);

    /**
     * broker响应返回之前执行
     */
    void doAfterResponse(final String remoteAddr, final RemotingCommand request,
        final RemotingCommand response);
}

3.4 多端口监听

Broker启动时, 会监听3个端口: 10911、10911、10912

remotingServer: 监听10911, 可以处理客户端的所有请求。
fastRemotingServer: 监听10909, 可以处理客户端除了拉取消息之外的所有请求, vip端口。
haListenPort: 监听10912, 于Broker的主从同步, 即高可用服务。

通过两个端口监听客户端请求, 达到隔离读写操作效果。消息的发送中, 重要的是提供RTT。如果普通端口堵塞, 会使得netty的IO线程阻塞, 比如消息堆积的时候, 消费消息的请求会填满IO线程池, 导致写操作被阻塞, 所以可以向vip端口发送消息。

现在发送消息和消费消息实际上默认都走10911端口。vip端口默认不开通, 需要配置属性, com.rocketmq.sendMessageWithVIPChannel属性。

4.Start启动BrokerController

创建和初始化Broker后, 启动Broker。

BrokerController#start

/**
 * BrokerStartup的方法
 *
 * @param controller 被启动的BrokerController
 */
public static BrokerController start(BrokerController controller) {
    try {
        //启动broker
        controller.start();

        //broker启动之后的的信息打印到控制台
        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;
}

启动各种服务和组件: messageStore, remotingServer, fastRemotingServer
判断如果没有开启DLedger, 如果当前broker不是SLAVE, 那么启动transactionalMessageCheckService事务消息检查服务, 如果是SLAVE, 那么启动定时任务每隔10s与master机器同步数据, 采用slave主动拉取的方法。
最后还会开启另一个定时周期任务, 默认情况下每隔30s向所有的nameServer进行一次注册broker信息, 时间间隔可以配置registerNameServerPeriod属性, 允许的值是在1万到6万毫秒之间。这个任务是Broker向nameServer发送心跳包的定时任务, 包含topic读、写队列个数、队列权限、是否有序等信息。

/**
 * BrokerController的方法
 * 启动BrokerController
 */
public void start() throws Exception {
    //启动消息存储服务
    if (this.messageStore != null) {
        this.messageStore.start();
    }
    //启动netty远程服务
    if (this.remotingServer != null) {
        this.remotingServer.start();
    }
    //启动快速netty远程服务
    if (this.fastRemotingServer != null) {
        this.fastRemotingServer.start();
    }
    //文件监听器启动
    if (this.fileWatchService != null) {
        this.fileWatchService.start();
    }
    //broker对外api启动
    if (this.brokerOuterAPI != null) {
        this.brokerOuterAPI.start();
    }
    //长轮询拉取消息挂起服务启动
    if (this.pullRequestHoldService != null) {
        this.pullRequestHoldService.start();
    }
    //客户端连接心跳服务启动
    if (this.clientHousekeepingService != null) {
        this.clientHousekeepingService.start();
    }
    //过滤服务管理器启动
    if (this.filterServerManager != null) {
        this.filterServerManager.start();
    }
    //如果没有开启DLeger的相关设置，默认没有启动
    if (!messageStoreConfig.isEnableDLegerCommitLog()) {
        //如果不是SLAVE，那么启动transactionalMessageCheckService事务消息检查服务
        startProcessorByHa(messageStoreConfig.getBrokerRole());
        //如果是SLAVE，那么启动定时任务每隔10s与master机器同步数据，采用slave主动拉取的方法
        //同步的内容包括topic配置，消费者消费位移、延迟消息偏移量、订阅组信息等
        handleSlaveSynchronize(messageStoreConfig.getBrokerRole());
        /*
         * 强制注册当前broker信息到所有的nameserver
         */
        this.registerBrokerAll(true, false, true);
    }
    //启动定时任务，默认情况下每隔30s向nameServer进行一次注册，
    //时间间隔可以配置registerNameServerPeriod属性，允许的值是在1万到6万毫秒之间。
    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);
            }
        }
    }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);
    //broker相关统计服务启动
    if (this.brokerStatsManager != null) {
        this.brokerStatsManager.start();
    }
    //broker快速失败服务启动
    if (this.brokerFastFailure != null) {
        this.brokerFastFailure.start();
    }
}