RocketMQ学习九-Broker初步分析 - HelloWorld开发者社区

一，Broker缓存的数据

Broker主要缓存了路由信息，包含producer表,consumer表,consumerGroup表和topic表。这些信息是在ProducerManager，ConsumerManager，SubscriptionGroupManager，TopicConfigManager这几个类里进行管理的。

ProducerManager
//producer列表
HashMap<String /* group name */, HashMap<Channel, ClientChannelInfo>> groupChannelTable

groupChannelTable:各ProducerGroup中分别有哪些存活的Producer连接；每个连接的Producer最后一次发来心跳的时间

ConsumerManager
//consumer列表
ConcurrentMap<String/* Group */, ConsumerGroupInfo> consumerTable

consumerTable:每个ConsumerGroup中分别有哪些存活的Consumer连接，分别订阅了哪些Topic，订阅的每个Topic使用什么过滤条件（TAG）。

SubscriptionGroupManager
//ConsumerGroup表
ConcurrentMap<String, SubscriptionGroupConfig> subscriptionGroupTable

subscriptionGroupTable:各ConsumerGroup的消费行为特点，例如：消费失败后的最大重试次数；重试队列个数；如果从MasterBroker消费缓慢，切换到哪个Slave Broker进行消费

TopicConfigManager
//topic列表
ConcurrentMap<String, TopicConfig> topicConfigTable

topicConfigTable:分布在当前Broker上的各Topic分片的配置信息，如：包含的读/写Queue的数量；是否有读/写权限

二，Broker启动设计

创建BrokerController
创建BrokerController类是在BrokerStartup#createBrokerController方法里进行的。先是进行参考解析，完了创建BrokerController类，紧接着调用其initialize方法，里面的逻辑主要有：
1）加载topic,consumer消费进度，订阅关系与consumer过滤的配置，并会加载消息的日志文件
2）再创建一个netty服务监听10909这个VIP端口
3）初始化一系列线程池，然后在registerProcessor方法里将这些线程池与处理器进行关联，为以后不同的业务使用不同的线程池，也就是线程隔离
4）启动一些定时任务，比如记录Broker状态，消费进度持久化等
5）最后进行权限校验初始化和Rpc调用钩子相关服务，这些服务加载方式是Java的SPI方式进行的。

启动Broker

 public void start() throws Exception {
     //启动消息存储相关的任务
     if (this.messageStore != null) {
         this.messageStore.start();
     }
     //启动broker服务器
     if (this.remotingServer != null) {
         this.remotingServer.start();
     }
     //启动给消息发送者使用的netty服务
     if (this.fastRemotingServer != null) {
         this.fastRemotingServer.start();
     }
     //启动监控SSL连接文件的服务
     if (this.fileWatchService != null) {
         this.fileWatchService.start();
     }
     //启动外部API的客户端
     if (this.brokerOuterAPI != null) {
         this.brokerOuterAPI.start();
     }
     //启动pull模式相关的服务
     if (this.pullRequestHoldService != null) {
         this.pullRequestHoldService.start();
     }
     //启动心跳检测服务
     if (this.clientHousekeepingService != null) {
         this.clientHousekeepingService.start();
     }
     //启动消息过滤服务
     if (this.filterServerManager != null) {
         this.filterServerManager.start();
     }
     //如果没启动DLegerCommitLog ，就将Broker注册到NameServer上
     if (!messageStoreConfig.isEnableDLegerCommitLog()) {
         startProcessorByHa(messageStoreConfig.getBrokerRole());
         handleSlaveSynchronize(messageStoreConfig.getBrokerRole());
     }

     /*向namesrv注册*/
     this.registerBrokerAll(true, false, true);

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

     if (this.brokerStatsManager != null) {
         this.brokerStatsManager.start();
     }

     if (this.brokerFastFailure != null) {
         this.brokerFastFailure.start();
     }


 }

messageStore服务：处理消息的存储相关的日志，比如CommitLog，ConsumeQueue等
remotingServer服务：处理客户端producer&consumer的请求
fastRemotingServer服务：默认端口可能存在多用，可能会造成业务阻塞。新开一个VIP端口专门进行消息处理。不过4.5版本之后默认已关闭，是为了妆容之前版本。
fileWatchService服务：启动监控服务连接时用到的SSL连接文件的服务
brokerOuterAPI服务：RocketMQ控制台跟Broker交互时候的客户端
pullRequestHoldService服务：处理push模式消费，或者延迟消费的服务
clientHousekeepingService服务：心跳连接用的服务
filterServerManager服务：过滤消息服务
transactionalMessageCheckService服务：定期检查和处理事务消息服务
slaveSynchronize服务：主从路由信息同步服务