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