1.rocketmq圖形化控制台安裝 雖然rocketmq為用戶提供了使用命令行管理主題、消費組以及broker配置的功能,但對於不夠熟練的非運維人員來說,命令行的管理界面還是較難使用的。為此,我們可以使用圖形化的管理界面來簡化管理操作。 rocketmq官方推薦的圖形化控制台目前還處在不成熟的孵化 ...
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外,也能夠從源碼中學習到許多架構設計和編程實踐相關的知識。
如有理解不到位的地方,請多多指教。