RocketMQ 原始碼分析 BrokerStartUp（七）
阿新 • • 發佈：2018-12-30
前言
broker是訊息接收處理，訊息落地的核心模組。這個模組用於接收producer傳送的訊息以及consumer消費的訊息。 本章從broker的啟動類入手分析一下broker啟動的過程。
分析broker是為了能夠搞明白master和slave之間的關係，以及訊息落地的具體過程。
createBrokerController
broker啟動會獲取外部傳入的初始化引數， 也會從配置檔案中載入相關的配置引數 ，將引數儲存在對應的Configuration類當中。
public static BrokerController createBrokerController 當配置資訊設定完畢後, broker會將這些資訊傳入brokerController控制器當中，這個控制器會初始化載入很多的管理器：
- rocketmq配置檔案資訊可以參考如下連結地址
- rocketMQ配置檔案資訊
- topicManager : 用於管理broker中儲存的所有topic的配置
- consumerOffsetManager: 管理Consumer的消費進度
- subscriptionGroupManager: 用來管理訂閱組，包括訂閱許可權等。
當管理器全部載入完畢後，控制器將開始進入下一步初始化，初始化多個執行緒池，包括sendMessageExecutor、pullMessageExecutor、adminBrokerExecutor、clientManagerExecutor, 分別用於傳送訊息執行器、拉取訊息執行器、broker管理器執行器、客戶端管理執行器， 這些執行器回放如執行緒池中處理, 來做併發執行。
public boolean initialize() throws CloneNotSupportedException {
//用於管理broker中儲存的所有topic的配置
boolean result = this.topicConfigManager.load();
//管理Consumer的消費進度
result = result && this.consumerOffsetManager.load();
//用來管理訂閱組，包括訂閱許可權等
result = result && this.subscriptionGroupManager.load();
// 用於消費者過濾配置
result = result && this.consumerFilterManager.load();
if (result) {//載入topicConfigManager、consumerOffsetManager、subscriptionGroupManager、consumerFilterManager ，將載入結果成功與否儲存在result中
try {
//用於broker層的訊息落地儲存
this.messageStore =
new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener,
this.brokerConfig);
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);
}
}
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);
this.fastRemotingServer = new NettyRemotingServer(fastConfig, this.clientHousekeepingService);
//傳送訊息的執行緒組
this.sendMessageExecutor = new BrokerFixedThreadPoolExecutor(
this.brokerConfig.getSendMessageThreadPoolNums(),
this.brokerConfig.getSendMessageThreadPoolNums(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.sendThreadPoolQueue,
new ThreadFactoryImpl("SendMessageThread_"));
//設定拉取訊息的執行緒組
this.pullMessageExecutor = new BrokerFixedThreadPoolExecutor(
this.brokerConfig.getPullMessageThreadPoolNums(),
this.brokerConfig.getPullMessageThreadPoolNums(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.pullThreadPoolQueue,
new ThreadFactoryImpl("PullMessageThread_"));
//設定查詢訊息的執行緒組
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_"));
//把剛剛建立的執行器註冊到 remotingServer, fastRemotingServer物件中
this.registerProcessor();
final long initialDelay = UtilAll.computNextMorningTimeMillis() - System.currentTimeMillis();
final long period = 1000 * 60 * 60 * 24;
//定期查詢如下資訊
//記錄broker的狀態
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);
//消費者當前資訊的offset位置
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void