rocketmq學習(二) rocketmq集群部署與圖形化控制台安裝

• 2019 年 10 月 3 日
• 筆記

1.rocketmq圖形化控制台安裝

　　雖然rocketmq為用戶提供了使用命令行管理主題、消費組以及broker配置的功能，但對於不夠熟練的非運維人員來說，命令行的管理界面還是較難使用的。為此，我們可以使用圖形化的管理界面來簡化管理操作。

　　rocketmq官方推薦的圖形化控制台目前還處在不成熟的孵化階段。倉庫地址為(https://github.com/apache/rocketmq-externals)，其中包含了rocketmq相關拓展的、屬於孵化期的各種項目。下載源碼之後，找到rocketmq-console文件夾，這就是rocketmq官方推薦的圖形化控制台項目，基於springboot和angularJS。

　　打開application.properties，能看到一些重要參數的配置，例如端口，nameServer地址，登錄權限控制等等。對於啟動參數的設置，可以選擇直接在配置文件中修改；也可在啟動項目時通過命令行指定。

　　為部署項目，先執行maven的打包命令(mvn clean package)，生成jar包。

　　然後執行java -jar rocketmq-console-ng-1.0.1.jar –rocketmq.config.namesrvAddr=localhost:9876(nameServer地址) –server.port=8080(啟動端口)。

　　通過瀏覽器訪問項目啟動的ip/port，即可看到以下管理界面(右上角可以中英文切換)。

 　　至此，rocketmq圖形化控制台安裝成功。

2.rocketmq集群部署

　　rocketmq的單機部署雖然簡單方便，卻存在着單點故障的問題。通過集群部署nameServer和broker可以實現rocketmq服務端的高可用。

　　下面介紹rocketmq的集群部署，以在兩台機器上搭建一個雙主雙從的高可用rocketmq集群為例子，這兩台機器(linux環境)的IP地址分別是192.168.32.130,和192.168.32.131。

2.1 nameServer集群部署

　　由於nameServer的集群節點之間互不通信，所以不需要額外的配置。

　　在兩台機器的rocketmq安裝的根路徑下分別執行“sh bin/mqnamesrv”，各啟動一個nameServer，使用其默認的端口9876。

　　現在，192.168.32.130:9876和192.168.32.131:9876上都各運行着一個nameServer服務。

2.2 broker集群部署

　　broker作為rocketmq的核心，其運行的穩定性至關重要。前面提到的雙主雙從(2-master-2-slave)實際上指代的是broker的集群運行模式，從broker作為主broker的備份，負責和主broker保持數據同步，可讀不可寫。

　　rocketmq通過賦予broker名稱來區別不同角色的broker。我們把當前兩個角色的broker分別命名為broker-a和broker-b。

　　在192.168.32.130 部署broker-a的主和broker-b的從，在192.168.32.131 部署broker-b的主和broker-a的從，同一角色的主從分別部署在不同的機器上。這樣，即使任意一台機器掛掉，由於從broker的存在，broker-a和broker-b依然可以對外提供服務。

　　啟動broker時，可以通過指定配置文件的方式為broker設置一系列參數。不同角色的broker，主從broker之間的配置文件參數內容各不相同。

broker配置文件中參數介紹：

　　brokerName：broker名稱，互為主從的broker名稱保持一致。

　　namesrvAddr：關聯的nameserver地址，多個用“;”隔開。

　　listenPort：broker監聽端口，同一機器部署多個broker不能監聽端口不能相同，避免衝突

　　storePathRootDir：broker存儲數據的根目錄

　　brokerClusterName：broker集群名稱，相同集群的master能互相識別

　　brokerId：0代表master，大於0代表不同的slave-broker

　　deleteWhen：刪除過時消息的時間，04代表凌晨4點

　　fileReservedTime：落盤數據文件保存的時長，單位小時

　　brokerRole：brokerRole有三種類型，SYNC_MASTER、ASYNC_MASTER和SLAVE，SYNC_M和ASYNC_M都代表主broker，區別在於主從之間進行數據同步的方式不同。SYNC代表主從數據同步完成，才向客戶端返回消息發送成功結果；反之ASYNC代表主庫收到消息後立即返回發送消息成功結果。

　　可以看到，ASYNC_MASTER的效率更高，但是當MASTER出現故障時，可能出現消息丟失的問題。需要用戶進行效率與可靠性之間的取捨。

　　flushDiskType：flushDiskType有兩種類型，SYNC_FLUSH(同步刷盤)和ASYNC_FLUSH(異步刷盤)，用於指定broker在接收到消息後，返回消息發送結果和數據落盤處理的策略。當選擇同步刷盤時，只有當消息數據真正的寫入磁盤持久化時，才返回消息發送成功。選擇異步刷盤時，消息數據寫入本地虛擬內存映射後，就直接返回。

　　broker的本地落盤策略和主從同步策略的選擇類似，都需要在效率與可靠性、一致性之間進行取捨。

　　比較推薦的一種配置方案是主從同步策略選擇SYNC_MASTER而本地落盤策略選擇SYNC_FLUSH。從可靠性的角度來看，只要主從broker沒有同時掛掉(避免了單點故障)，消息將不會丟失；從效率的角度來看，由於主從broker都是異步落盤，執行效率也有一定的保障，是一個優秀的折中方案。

broker配置文件詳情：

broker-a主broker  配置文件(broker-a.properties)：

brokerName=broker-a  namesrvAddr=192.168.32.130:9876;192.168.32.131:9876  listenPort=10911  storePathRootDir=/root/rocketmq/data/store/store-a  brokerClusterName=DefaultCluster  brokerId=0  deleteWhen=04  fileReservedTime=48  brokerRole=SYNC_MASTER  flushDiskType=ASYNC_FLUSH

broker-b主broker  配置文件(broker-b.properties)：

brokerName=broker-b  namesrvAddr=192.168.32.130:9876;192.168.32.131:9876  listenPort=10911  storePathRootDir=/root/rocketmq/data/store/store-b  brokerClusterName=DefaultCluster  brokerId=0  deleteWhen=04  fileReservedTime=48  brokerRole=SYNC_MASTER  flushDiskType=ASYNC_FLUSH

broker-a從broker  配置文件(broker-a-s.properties)：

brokerName=broker-a  namesrvAddr=192.168.32.130:9876;192.168.32.131:9876  listenPort=11011  storePathRootDir=/root/rocketmq/data/store/store-a  brokerClusterName=DefaultCluster  brokerId=1  deleteWhen=04  fileReservedTime=48  brokerRole=SLAVE  flushDiskType=ASYNC_FLUSH

broker-b從broker  配置文件(broker-b-s.properties)：

brokerName=broker-b  namesrvAddr=192.168.32.130:9876;192.168.32.131:9876  listenPort=11011  storePathRootDir=/root/rocketmq/data/store/store-b  brokerClusterName=DefaultCluster  brokerId=1  deleteWhen=04  fileReservedTime=48  brokerRole=SLAVE  flushDiskType=ASYNC_FLUSH

 　　在對應機器依次執行以下命令，依次啟動broker(先啟動主broker，後啟動從broker)。

　　192.168.32.130執行：sh bin/mqbroker -c [配置文件路徑 eg: rocketmq/data/conf/broker-a.properties]

　　192.168.32.131執行：sh bin/mqbroker -c [配置文件路徑 eg: rocketmq/data/conf/broker-b.properties]

　　192.168.32.131執行：sh bin/mqbroker -c [配置文件路徑 eg: rocketmq/data/conf/broker-a-s.properties]

　　192.168.32.130執行：sh bin/mqbroker -c [配置文件路徑 eg: rocketmq/data/conf/broker-b-s.properties]

　　此時，rocketmq雙主雙從的broker集群已經搭建完畢。啟動圖形控制台，指定命令行參數namesrvAddr=192.168.32.130:9876;192.168.32.131:9876，可看到以下信息：

3. 使用java客戶端收發消息

　　前面我們通過命令行的方式進行了rocketmq收發消息的測試。但在實際使用過程中，還是需要sdk客戶端來進行收發消息。這裡，我們使用rocketmq提供的java sdk來進行rocketmq的消息收發實驗。

　　先通過圖形控制台創建一個主題用於測試，主題名稱為”TopicTest”(隨便取的)。

　　接着啟動一個java項目，加入rocketmq-client的依賴。

maven坐標：

<!-- 原生rocketmq client -->  <dependency>       <groupId>org.apache.rocketmq</groupId>       <artifactId>rocketmq-client</artifactId>       <version>4.4.0</version>  </dependency>

生產者producer示例代碼：

public class Producer {      public static void main(String[] args) throws Exception {          final DefaultMQProducer producer = new DefaultMQProducer("test_producer_group");          // 設置nameServer地址          producer.setNamesrvAddr("192.168.32.130:9876;192.168.32.131:9876");          producer.start();          for (int i = 0; i < 1000; i++) {              try {                  // 構造消息對象，topic=TopicTest，tag=TagA                  Message msg = new Message("TopicTest", "TagA" ,                      ("Hello RocketMQ TopicTest" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));                  // 發送消息                  SendResult sendResult = producer.send(msg);                  System.out.printf("%s%n", sendResult);              } catch (Exception e) {                  e.printStackTrace();                  Thread.sleep(1000);              }          }          producer.shutdown();      }  }

消費者consumer示例代碼：

public class Consumer {      public static void main(String[] args) throws Exception {          DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_consumer_group");          // 設置nameServer地址          consumer.setNamesrvAddr("192.168.32.130:9876;192.168.32.131:9876");          consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);            // 訂閱主題 tag="*"代表訂閱TopicTest主題下所有子主題消息          consumer.subscribe("TopicTest", "*");          // 註冊消息監聽回調函數          consumer.registerMessageListener((MessageListenerConcurrently)(msgs, context) -> {              for(MessageExt messageExt : msgs){                  String strMsg = new String(messageExt.getBody());                  System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), strMsg);              }              return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;          });          consumer.start();          System.out.printf("Consumer Started.%n");      }  }

　　生產者和消費者都是簡單的main方法啟動，先啟動producer發送消息，然後啟動consumer接受消息，控制台上將會看到接受時消息的日誌。可以試試主動關閉一個master-broker，看看broker集群的消息收發是否正常。

　　至此，通過java客戶端使用rocketmq的測試告一段落。

總結

　　本篇博客介紹了rocketmq的集群部署，圖形化界面的安裝以及如何使用java客戶端與rocketmq進行交互。rocketmq還有着許多好用，強大的功能，後續的博客將結合著rocketmq的源碼來介紹它們。

　　去閱讀並理解源碼，可以在解決問題時能看得更深、更遠。通過閱讀rocketmq的源碼，除了更好地掌握rocketmq外，也能夠從源碼中學習到許多架構設計和編程實踐相關的知識。

　　如有理解不到位的地方，請多多指教。