上周花了一點時間從頭到尾、從無到有地搭建了一套RocketMQ的環境，覺得還挺easy的，所以就寫篇文章分享給大家。
前言RocketMQ是阿里巴巴旗下一款開源的MQ框架，經歷過雙十一考驗、Java編程語言實現，有非常好完整生態系統。RocketMQ作為一款純java、分布式、隊列模型的開源消息中間件，支持事務消息、順序消息、批量消息、定時消息、消息回溯等，總之就是葛大爺的一句話
整篇文章可以大致分為三個部分，第一部分屬於一些核心概念和工作流程的講解；第二部分就是純手動搭建了一套環境；第三部分是基於環境進行測試和集成到SpringBoot，因為整個過程講的比較細，所以我稱之為「保姆級教程」。
好了，廢話補多少，直接進入主題。
核心概念NameServer：可以理解為是一個註冊中心，主要是用來保存topic路由信息，管理Broker。在NameServer的集群中，NameServer與NameServer之間是沒有任何通信的。
Broker：核心的一個角色，主要是用來保存topic的信息，接受生產者產生的消息，持久化消息。在一個Broker集群中，相同的BrokerName可以稱為一個Broker組，一個Broker組中,BrokerId為0的為主節點，其它的為從節點。BrokerName和BrokerId是可以在Broker啟動時通過配置文件配置的。每個Broker組只存放一部分消息。
生產者：生產消息的一方就是生產者
生產者組：一個生產者組可以有很多生產者，只需要在創建生產者的時候指定生產者組，那麼這個生產者就在那個生產者組
消費者：用來消費生產者消息的一方
消費者組：跟生產者一樣，每個消費者都有所在的消費者組，一個消費者組可以有很多的消費者，不同的消費者組消費消息是互不影響的。
topic（主題）：可以理解為一個消息的集合的名字，生產者在發送消息的時候需要指定發到哪個topic下，消費者消費消息的時候也需要知道自己消費的是哪些topic底下的消息。
Tag（子主題）：比topic低一級，可以用來區分同一topic下的不同業務類型的消息，發送消息的時候也需要指定。
這裡有組的概念是因為可以用來做到不同的生產者組或者消費者組有不同的配置，這樣就可以使得生產者或者消費者更加靈活。
工作流程說完核心概念，再來說一下核心的工作流程，這裡我先畫了一張圖。
通過這張圖就可以很清楚的知道，RocketMQ大致的工作流程：
Broker啟動的時候，會往每台NameServer（因為NameServer之間不通信，所以每台都得註冊）註冊自己的信息，這些信息包括自己的ip和端口號，自己這台Broker有哪些topic等信息。
Producer在啟動之後會跟會NameServer建立連接，定期從NameServer中獲取Broker的信息，當發送消息的時候，會根據消息需要發送到哪個topic去找對應的Broker地址，如果有的話，就向這台Broker發送請求；沒有找到的話，就看根據是否允許自動創建topic來決定是否發送消息。
Broker在接收到Producer的消息之後，會將消息存起來，持久化，如果有從節點的話，也會主動同步給從節點，實現數據的備份
Consumer啟動之後也會跟會NameServer建立連接，定期從NameServer中獲取Broker和對應topic的信息，然後根據自己需要訂閱的topic信息找到對應的Broker的地址，然後跟Broker建立連接，獲取消息，進行消費
就跟上面的圖一樣，整體的工作流程還是比較簡單的，這裡我簡化了很多概念，主要是為了好理解。
環境搭建終於講完了一些簡單的概念，接下來就來搭建一套RocketMQ的環境。
通過上面分析，我們知道，在RocketMQ中有NameServer、Broker、生產者、消費者四種角色。而生產者和消費者實際上就是業務系統，所以這裡不需要搭建，真正要搭建的就是NameServer和Broker，但是為了方便RocketMQ數據的可視化，這裡我多搭建一套可視化的服務。
搭建過程比較簡單，按照步驟一步一步來就可以完成，如果提示一些命令不存在，那麼直接通過yum安裝這些命令就行。
一、準備需要準備一個linux服務器，需要先安裝好JDK
關閉防火牆
systemctlstopfirewalldsystemctldisablefirewalld下載並解壓RocketMQ1、創建一個目錄，用來存放rocketmq相關的東西mkdir/usr/rocketmqcd/usr/rocketmq2、下載並解壓rocketmq下載
wgethttps://archive.apache.org/dist/rocketmq/4.7.1/rocketmq-all-4.7.1-bin-release.zip解壓
unziprocketmq-all-4.7.1-bin-release.zip看到這一個文件夾就完成了
然後進入rocketmq-all-4.7.1-bin-release文件夾
cdrocketmq-all-4.7.1-bin-releaseRocketMQ的東西都在這了
二、搭建NameServer修改jvm參數在啟動NameServer之前，強烈建議修改一下啟動時的jvm參數，因為默認的參數都比較大，為了避免內存不夠，建議修改小，當然，如果你的內存足夠大，可以忽略。
vibin/runserver.sh修改畫圈的這一行
這裡你可以直接修改成跟我一樣的
-server -Xms512m -Xmx512m -Xmn256m -XX:MetaspaceSize=32m -XX:MaxMetaspaceSize=50m啟動NameServer修改完之後，執行如下命令就可以啟動NameServer了
nohupshbin/mqnamesrv&查看NameServer日誌
tail-f~/logs/rocketmqlogs/namesrv.log如果看到如下的日誌，就說明啟動成功了
NameServer日誌三、搭建Broker這裡啟動單機版的Broker
修改jvm參數跟啟動NameServer一樣，也建議去修改jvm參數
vibin/runbroker.sh將畫圈的地方設置小點，當然也別太小啊
當然你還是可以跟我設置的一樣
-server-Xms1g-Xmx1g-Xmn512m修改Broker配置文件broker.conf這裡需要改一下Broker配置文件，需要指定NameServer的地址，因為需要Broker需要往NameServer註冊
viconf/broker.confBroker配置文件
Broker配置文件這裡就能看出Broker的配置了，什麼Broker集群的名稱啊，Broker的名稱啊，Broker的id啊，都跟前面說的對上了。
在文件末尾追加地址
namesrvAddr=localhost:9876因為NameServer跟Broker在同一台機器，所以是localhost，NameServer端口默認的是9876。
不過這裡我還建議再修改一處信息，因為Broker向NameServer進行註冊的時候，帶過去的ip如果不指定就會自動獲取，但是自動獲取的有個坑，就是有可能你的電腦無法訪問到這個自動獲取的ip，所以我建議手動指定你的電腦可以訪問到的服務器ip。
我的虛擬機的ip是192.168.200.143，所以就指定為192.168.200.143，如下
brokerIP1=192.168.200.143brokerIP2=192.168.200.143如果以上都配置的話，最終的配置文件應該如下，紅圈的為新加的
啟動Brokernohupshbin/mqbroker-cconf/broker.conf&-c 參數就是指定配置文件
查看日誌
tail-f~/logs/rocketmqlogs/broker.log當看到如下日誌就說明啟動成功了
四、搭建可視化控制台其實前面NameServer和Broker搭建完成之後，就可以用來收發消息了，但是為了更加直觀，可以搭一套可視化的服務。
可視化服務其實就是一個jar包，啟動就行了。
jar包可以從這獲取
鏈接：https://pan.baidu.com/s/16s1qwY2qzE2bxR81t5Wm6w提取碼：s0sd將jar包上傳到服務器，放到/usr/rocketmq的目錄底下，當然放哪都無所謂，這裡只是為了方便，因為rocketmq的東西都在這裡
然後進入/usr/rocketmq下，執行如下命名
nohupjava-jar-server-Xms256m-Xmx256m-Drocketmq.config.namesrvAddr=localhost:9876-Dserver.port=8088rocketmq-console-ng-1.0.1.jar&rocketmq.config.namesrvAddr就是用來指定NameServer的地址的
查看日誌
tail-f~/logs/consolelogs/rocketmq-console.log當看到如下日誌，就說明啟動成功了
然後在瀏覽器中輸入http://linux服務器的ip:8088/就可以看到控制台了，如果無法訪問，可以看看防火牆有沒有關閉
右上角可以把語言切換成中文
Broker集群信息
topic信息通過控制台可以查看生產者、消費者、Broker集群等信息，非常直觀。
功能很多，這裡就不一一介紹了。
測試環境搭好之後，就可以進行測試了。
引入依賴<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.7.1</version></dependency>生產者發送消息publicclassProducer{publicstaticvoidmain(String[]args)throwsException{//創建一個生產者，指定生產者組為sanyouProducerDefaultMQProducerproducer=newDefaultMQProducer("sanyouProducer");//指定NameServer的地址producer.setNamesrvAddr("192.168.200.143:9876");//第一次發送可能會超時，我設置的比較大producer.setSendMsgTimeout(60000);//啟動生產者producer.start();//創建一條消息//topic為sanyouTopic//消息內容為三友的java日記//tags為TagAMessagemsg=newMessage("sanyouTopic","TagA","三友的java日記".getBytes(RemotingHelper.DEFAULT_CHARSET));//發送消息並得到消息的發送結果，然後打印SendResultsendResult=producer.send(msg);System.out.printf("%s%n",sendResult);//關閉生產者producer.shutdown();}}運行結果如下
SendResult[sendStatus=SEND_OK,msgId=C0A81FAF54F818B4AAC2475FD2010000,offsetMsgId=C0A8C88F00002A9F000000000009AE55,messageQueue=MessageQueue[topic=sanyouTopic,brokerName=broker-a,queueId=0],queueOffset=0]sendStatus=SEND_OK 說明發送成功了，此時就能後控制台看到未消費的消息了。
到控制台看到消息那塊，然後選定發送的topic，查詢的時間範圍手動再選一下，不選就查不出來(我懷疑這是個bug)，然後查詢就能看到了一條消息。
然後點擊一下MESSAGE DETAIL就能夠看到詳情。
這裡就能看到發送消息的詳細信息。
左下角消息的消費的消費，因為我們還沒有消費者訂閱這個topic，所以左下角沒數據。
消費者消費消息publicclassConsumer{publicstaticvoidmain(String[]args)throwsInterruptedException,MQClientException{//通過push模式消費消息，指定消費者組DefaultMQPushConsumerconsumer=newDefaultMQPushConsumer("sanyouConsumer");//指定NameServer的地址consumer.setNamesrvAddr("192.168.200.143:9876");//訂閱這個topic下的所有的消息consumer.subscribe("sanyouTopic","*");//註冊一個消費的監聽器，當有消息的時候，會回調這個監聽器來消費消息consumer.registerMessageListener(newMessageListenerConcurrently(){@OverridepublicConsumeConcurrentlyStatusconsumeMessage(List<MessageExt>msgs,ConsumeConcurrentlyContextcontext){for(MessageExtmsg:msgs){System.out.printf("消費消息:%s",newString(msg.getBody())+"\n");}returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});//啟動消費者consumer.start();System.out.printf("ConsumerStarted.%n");}}啟動之後，消費者就會消費剛才生產者發送的消息，於是控制台就打印出如下信息
ConsumerStarted.消費消息:三友的java日記此時再去看控制台
發現被sanyouConsumer這個消費者組給消費了。
SpringBoot環境下集成RocketMQ 集成在實際項目中肯定不會像上面測試那樣用，都是集成SpringBoot的。
1、引入依賴<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.1.1</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><version>2.1.1.RELEASE</version></dependency>2、yml配置rocketmq:producer:group:sanyouProducername-server:192.168.200.143:98763、創建消費者SpringBoot底下只需要實現RocketMQListener接口，然後加上@RocketMQMessageListener註解即可
@Component@RocketMQMessageListener(consumerGroup="sanyouConsumer",topic="sanyouTopic")publicclassSanYouTopicListenerimplementsRocketMQListener<String>{@OverridepublicvoidonMessage(Stringmsg){System.out.println("處理消息:"+msg);}}@RocketMQMessageListener需要指定消費者屬於哪個消費者組，消費哪個topic，NameServer的地址已經通過yml配置文件配置類
4、測試@SpringBootTest(classes=RocketMQApplication.class)@RunWith(SpringRunner.class)publicclassRocketMQTest{@AutowiredprivateRocketMQTemplatetemplate;@Testpublicvoidsend()throwsInterruptedException{template.convertAndSend("sanyouTopic","三友的java日記");Thread.sleep(60000);}}直接注入一個RocketMQTemplate，然後通過RocketMQTemplate發送消息。
運行結果如下：
處理消息:三友的java日記的確消費到消息了。
原理其實原理是一樣的，只不過在SpringBoot中給封裝了一層，讓使用起來更加簡單。
1、RocketMQTemplate構造代碼
所以從這可以看出，最終在構造RocketMQTemplate的時候，傳入了一個DefaultMQProducer，所以可想而知，最終RocketMQTemplate發送消息也是通過DefaultMQProducer發送的。
2、@RocketMQMessageListener 註解處理
從這可以看出，會為每一個加了@RocketMQMessageListener註解的對象創建一個DefaultMQPushConsumer，所以最終也是通過DefaultMQPushConsumer消費消息的。
至於監聽器，是在這
遍歷每條消息，然後調用handleMessage，最終會調用實現了RocketMQListener的對象處理消息。
最後通過上面的理論介紹和實際的環境搭建再加上代碼的測試，相信應該可以對RocketMQ有個入門，有興趣的小夥伴可以手動搭起來，整個過程順利的話可能就十幾二十分鐘這樣子。
最後我再說一句，從文章整體也可以看出本文沒有涉及太深入的一些機制和原理的講解，比如消息是如何存儲的，事務和延遲消息是如何實現的，主從是如何同步的等等，甚至壓根就沒提到隊列這個詞，主要是因為我打算後面再寫一篇文章，來單獨剖析這些機制和原理。
最後的最後，本文所有的代碼地址:
https://github.com/sanyou3/rocketmq-demo.git推薦
Java面試題寶典
技術內卷群，一起來學習！！
PS：因為公眾號平台更改了推送規則，如果不想錯過內容，記得讀完點一下「在看」，加個「星標」，這樣每次新文章推送才會第一時間出現在你的訂閱列表里。點「在看」支持我們吧！
留言列表