深入剖析RocketMQ源碼-NameServer
[導(dǎo)讀]作者：vivo互聯(lián)網(wǎng)服務(wù)器團(tuán)隊(duì)-YeWenhao一、RocketMQ架構(gòu)簡介1.1邏輯部署圖（圖片來自網(wǎng)絡(luò)）1.2核心組件說明通過上圖可以看到，RocketMQ的核心組件主要包括4個(gè)，分別是NameServer、Broker、Producer和Consumer，下面我們先依次簡單...
一、RocketMQ架構(gòu)簡介
1.1 邏輯部署圖
（圖片來自網(wǎng)絡(luò)）1.2 核心組件說明
通過上圖可以看到，RocketMQ的核心組件主要包括4個(gè)，分別是NameServer、Broker、Producer和Consumer，下面我們先依次簡單說明下這四個(gè)核心組件：
NameServer：NameServer充當(dāng)路由信息的提供者。生產(chǎn)者或消費(fèi)者能夠通過NameServer查找各Topic相應(yīng)的Broker IP列表。多個(gè)Namesrver實(shí)例組成集群，但相互獨(dú)立，沒有信息交換。
Broker：消息中轉(zhuǎn)角色，負(fù)責(zé)存儲(chǔ)消息、轉(zhuǎn)發(fā)消息。Broker服務(wù)器在RocketMQ系統(tǒng)中負(fù)責(zé)接收從生產(chǎn)者發(fā)送來的消息并存儲(chǔ)、同時(shí)為消費(fèi)者的拉取請(qǐng)求作準(zhǔn)備。Broker服務(wù)器也存儲(chǔ)消息相關(guān)的元數(shù)據(jù)，包括消費(fèi)者組、消費(fèi)進(jìn)度偏移和主題和隊(duì)列消息等。
Producer：負(fù)責(zé)生產(chǎn)消息，一般由業(yè)務(wù)系統(tǒng)負(fù)責(zé)生產(chǎn)消息。一個(gè)消息生產(chǎn)者會(huì)把業(yè)務(wù)應(yīng)用系統(tǒng)里產(chǎn)生的消息發(fā)送到Broker服務(wù)器。RocketMQ提供多種發(fā)送方式，同步發(fā)送、異步發(fā)送、順序發(fā)送、單向發(fā)送。同步和異步方式均需要Broker返回確認(rèn)信息，單向發(fā)送不需要。
Consumer：負(fù)責(zé)消費(fèi)消息，一般是后臺(tái)系統(tǒng)負(fù)責(zé)異步消費(fèi)。一個(gè)消息消費(fèi)者會(huì)從Broker服務(wù)器拉取消息、并將其提供給應(yīng)用程序。從用戶應(yīng)用的角度而言提供了兩種消費(fèi)形式：拉取式消費(fèi)、推動(dòng)式消費(fèi)。
除了上面說的三個(gè)核心組件外，還有Topic這個(gè)概念下面也會(huì)多次提到：
Topic：表示一類消息的集合，每個(gè)Topic包含若干條消息，每條消息只能屬于一個(gè)Topic，是RocketMQ進(jìn)行消息訂閱的基本單位。一個(gè)Topic可以分片在多個(gè)Broker集群上，每一個(gè)Topic分片包含多個(gè)queue，具體結(jié)構(gòu)可以參考下圖：
1.3 設(shè)計(jì)理念
RocketMQ是基于主題的發(fā)布與訂閱模式，核心功能包括消息發(fā)送、消息存儲(chǔ)、消息消費(fèi)，整體設(shè)計(jì)追求簡單與性能第一，歸納來說主要是下面三種：
- NameServer取代ZK充當(dāng)注冊(cè)中心，NameServer集群間互不通信，容忍路由信息在集群內(nèi)分鐘級(jí)不一致，更加輕量級(jí)；
- 使用內(nèi)存映射機(jī)制實(shí)現(xiàn)高效的IO存儲(chǔ)，達(dá)到高吞吐量；
- 容忍設(shè)計(jì)缺陷，通過ACK確保消息至少消費(fèi)一次，但是如果ACK丟失，可能消息重復(fù)消費(fèi)，這種情況設(shè)計(jì)上允許，交給使用者自己保證。
本文重點(diǎn)介紹的就是NameServer，我們下面一起來看下NameServer是如何啟動(dòng)以及如何進(jìn)行路由管理的。
二、NameServer架構(gòu)設(shè)計(jì)
在第一章已經(jīng)簡單介紹了NameServer取代zk作為一種更輕量級(jí)的注冊(cè)中心充當(dāng)路由信息的提供者。那么具體是如何來實(shí)現(xiàn)路由信息管理的呢？我們先看下圖：
上面的圖描述了NameServer進(jìn)行路由注冊(cè)、路由剔除和路由發(fā)現(xiàn)的核心原理。
路由注冊(cè)：Broker服務(wù)器在啟動(dòng)的時(shí)候會(huì)想NameServer集群中所有的NameServer發(fā)送心跳信號(hào)進(jìn)行注冊(cè)，并會(huì)每隔30秒向nameserver發(fā)送心跳，告訴NameServer自己活著。NameServer接收到Broker發(fā)送的心跳包之后，會(huì)記錄該broker信息，并保存最近一次收到心跳包的時(shí)間。
路由剔除：NameServer和每個(gè)Broker保持長連接，每隔30秒接收Broker發(fā)送的心跳包，同時(shí)自身每個(gè)10秒掃描BrokerLiveTable，比較上次收到心跳時(shí)間和當(dāng)前時(shí)間比較是否大于120秒，如果超過，那么認(rèn)為Broker不可用，剔除路由表中該Broker相關(guān)信息。
路由發(fā)現(xiàn)：路由發(fā)現(xiàn)不是實(shí)時(shí)的，路由變化后，NameServer不主動(dòng)推給客戶端，等待producer定期拉取最新路由信息。這樣的設(shè)計(jì)方式降低了NameServer實(shí)現(xiàn)的復(fù)雜性，當(dāng)路由發(fā)生變化時(shí)通過在消息發(fā)送端的容錯(cuò)機(jī)制來保證消息發(fā)送的高可用（這塊內(nèi)容會(huì)在后續(xù)介紹producer消息發(fā)送時(shí)介紹，本文不展開講解）。
高可用：NameServer通過部署多臺(tái)NameServer服務(wù)器來保證自身的高可用，同時(shí)多個(gè)NameServer服務(wù)器之間不進(jìn)行通信，這樣路由信息發(fā)生變化時(shí)，各個(gè)NameServer服務(wù)器之間數(shù)據(jù)可能不是完全相同的，但是通過發(fā)送端的容錯(cuò)機(jī)制保證消息發(fā)送的高可用。這個(gè)也正是NameServer追求簡單高效的目的所在。
三、?啟動(dòng)流程
在整理了解了NameServer的架構(gòu)設(shè)計(jì)之后，我們先來看下NameServer到底是如何啟動(dòng)的呢？
既然是源碼解讀，那么我們先來看下代碼入口：org.apache.rocketmq.namesrv.NamesrvStartup#main(String[] args)，實(shí)際調(diào)用的是main0()方法，代碼如下：
public static NamesrvController main0(String[] args) {
try { //創(chuàng)建namesrvController NamesrvController controller = createNamesrvController(args); //初始化并啟動(dòng)NamesrvController start(controller); String tip = "The Name Server boot success. serializeType=" RemotingCommand.getSerializeTypeConfigInThisServer(); log.info(tip); System.out.printf("%s%n", tip); return controller; } catch (Throwable e) { e.printStackTrace(); System.exit(-1); }
return null;}通過main方法啟動(dòng)NameServer，主要分為兩大步，先創(chuàng)建NamesrvController，然后再初始化并啟動(dòng)NamesrvController。我們分別展開來分析。
3.1 時(shí)序圖
具體展開閱讀代碼之前，我們先通過一個(gè)序列圖對(duì)整體流程有個(gè)了解，如下圖：
3.2 創(chuàng)建NamesrvController
先來看核心代碼，如下：
public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException { // 設(shè)置版本號(hào)為當(dāng)前版本號(hào) System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION)); //PackageConflictDetect.detectFastjson(); //構(gòu)造org.apache.commons.cli.Options,并添加-h -n參數(shù)，-h參數(shù)是打印幫助信息，-n參數(shù)是指定namesrvAddr Options options = ServerUtil.buildCommandlineOptions(new Options()); //初始化commandLine，并在options中添加-c -p參數(shù)，-c指定nameserver的配置文件路徑，-p標(biāo)識(shí)打印配置信息 commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser()); if (null == commandLine) { System.exit(-1); return null; } //nameserver配置類，業(yè)務(wù)參數(shù) final NamesrvConfig namesrvConfig = new NamesrvConfig(); //netty服務(wù)器配置類，網(wǎng)絡(luò)參數(shù) final NettyServerConfig nettyServerConfig = new NettyServerConfig(); //設(shè)置nameserver的端口號(hào) nettyServerConfig.setListenPort(9876); //命令帶有-c參數(shù)，說明指定配置文件，需要根據(jù)配置文件路徑讀取配置文件內(nèi)容，并將文件中配置信息賦值給NamesrvConfig和NettyServerConfig if (commandLine.hasOption('c')) { String file = commandLine.getOptionValue('c'); if (file != null) { InputStream in = new BufferedInputStream(new FileInputStream(file)); properties = new Properties(); properties.load(in); //反射的方式 MixAll.properties2Object(properties, namesrvConfig); MixAll.properties2Object(properties, nettyServerConfig); //設(shè)置配置文件路徑 namesrvConfig.setConfigStorePath(file);
System.out.printf("load config properties file OK, %s%n", file); in.close(); } } //命令行帶有-p，說明是打印參數(shù)的命令，那么就打印出NamesrvConfig和NettyServerConfig的屬性。在啟動(dòng)NameServer時(shí)可以先使用./mqnameserver -c configFile -p打印當(dāng)前加載的配置屬性 if (commandLine.hasOption('p')) { InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME); MixAll.printObjectProperties(console, namesrvConfig); MixAll.printObjectProperties(console, nettyServerConfig); //打印參數(shù)命令不需要啟動(dòng)nameserver服務(wù)，只需要打印參數(shù)即可 System.exit(0); } //解析命令行參數(shù)，并加載到namesrvConfig中 MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig); //檢查ROCKETMQ_HOME，不能為空 if (null == namesrvConfig.getRocketmqHome()) { System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation%n", MixAll.ROCKETMQ_HOME_ENV); System.exit(-2); } //初始化logback日志工廠，rocketmq默認(rèn)使用logback作為日志輸出 LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory(); JoranConfigurator configurator = new JoranConfigurator(); configurator.setContext(lc); lc.reset(); configurator.doConfigure(namesrvConfig.getRocketmqHome() "/conf/logback_namesrv.xml");
log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
MixAll.printObjectProperties(log, namesrvConfig); MixAll.printObjectProperties(log, nettyServerConfig); //創(chuàng)建NamesrvController final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
//將全局Properties的內(nèi)容復(fù)制到NamesrvController.Configuration.allConfigs中 // remember all configs to prevent discard controller.getConfiguration().registerConfig(properties);
return controller;}通過上面對(duì)每一行代碼的注釋，可以看出來，創(chuàng)建NamesrvController的過程主要分為兩步：
Step1：通過命令行中獲取配置。賦值給NamesrvConfig和NettyServerConfig類。Step2：根據(jù)配置類NamesrvConfig和NettyServerConfig構(gòu)造一個(gè)NamesrvController實(shí)例。
可見NamesrvConfig和NettyServerConfig是想當(dāng)重要的，這兩個(gè)類分別是NameServer的業(yè)務(wù)參數(shù)和網(wǎng)絡(luò)參數(shù)，我們分別看下這兩個(gè)類里面有哪些屬性：
NamesrvConfig
NettyServerConfig
注：Apache Commons CLI是開源的命令行解析工具，它可以幫助開發(fā)者快速構(gòu)建啟動(dòng)命令，并且?guī)椭憬M織命令的參數(shù)、以及輸出列表等。
3.3?初始化并啟動(dòng)
創(chuàng)建了NamesrvController實(shí)例之后，開始初始化并啟動(dòng)NameServer。
首先進(jìn)行初始化，代碼入口是NamesrvController#initialize。
public boolean initialize() { //加載kvConfigPath下kvConfig.json配置文件里的KV配置，然后將這些配置放到KVConfigManager#configTable屬性中 this.kvConfigManager.load(); //根據(jù)nettyServerConfig初始化一個(gè)netty服務(wù)器。 //brokerHousekeepingService是在NamesrvController實(shí)例化時(shí)構(gòu)造函數(shù)里實(shí)例化的，該類負(fù)責(zé)Broker連接事件的處理，實(shí)現(xiàn)了ChannelEventListener，主要用來管理RouteInfoManager的brokerLiveTable this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService); //初始化負(fù)責(zé)處理Netty網(wǎng)絡(luò)交互數(shù)據(jù)的線程池，默認(rèn)線程數(shù)是8個(gè) this.remotingExecutor = Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_")); //注冊(cè)Netty服務(wù)端業(yè)務(wù)處理邏輯，如果開啟了clusterTest，那么注冊(cè)的請(qǐng)求處理類是ClusterTestRequestProcessor，否則請(qǐng)求處理類是DefaultRequestProcessor this.registerProcessor(); //注冊(cè)心跳機(jī)制線程池，延遲5秒啟動(dòng)，每隔10秒遍歷RouteInfoManager#brokerLiveTable這個(gè)屬性，用來掃描不存活的broker this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override public void run() { NamesrvController.this.routeInfoManager.scanNotActiveBroker(); } }, 5, 10, TimeUnit.SECONDS); //注冊(cè)打印KV配置線程池，延遲1分鐘啟動(dòng)、每10分鐘打印出kvConfig配置 this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override public void run() { NamesrvController.this.kvConfigManager.printAllPeriodically(); } }, 1, 10, TimeUnit.MINUTES); //rocketmq可以通過開啟TLS來提高數(shù)據(jù)傳輸?shù)陌踩?，如果開啟了，那么需要注冊(cè)一個(gè)監(jiān)聽器來重新加載SslContext 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 
下載文檔
