RocketMQ入門到入土（一）新手也能看懂的原理和實戰！

阿新 • • 發佈：2020-06-28

學任何技術都是兩步驟：

搭建環境
helloworld

我也不例外，直接搞起來。

一、RocketMQ的安裝

1、檔案

官方網站

http://rocketmq.apache.org

GitHub

https://github.com/apache/rocketmq

2、下載

wgethttps://mirror.bit.edu.cn/apache/rocketmq/4.7.0/rocketmq-all-4.7.0-bin-release.zip

我們是基於Centos8來的，面向官方檔案學習，所以下載地址自然也是官方的。

去官方網站找合適的版本進行下載，目前我這裡最新的是4.7.0版本。

http://rocketmq.apache.org/dowloading/releases/

https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.7.0/rocketmq-all-4.7.0-bin-release.zip

3、準備工作

3.1、解壓

unziprocketmq-all-4.7.0-bin-release.zip

3.2、安裝jdk

sudoyuminstalljava-1.8.0-openjdk-devel

4、啟動

4.1、啟動namesrv

cdrocketmq-all-4.7.0-bin-release/bin
./mqnamesrv

4.2、啟動broker

cdrocketmq-all-4.7.0-bin-release/bin
 
./mqbroker-nlocalhost:9876

常見錯誤以及解決方案：

常見錯誤：啟動broker失敗Cannot allocate memory

[[email protected]bin]#./mqbroker-nlocalhost:9876
JavaHotSpot(TM)64-BitServerVMwarning:INFO:os::commit_memory(0x00000005c0000000,8589934592,0)failed
;error='Cannotallocatememory'(errno=12)#
#ThereisinsufficientmemoryfortheJavaRuntimeEnvironmenttocontinue.
 
#Nativememoryallocation(mmap)failedtomap8589934592bytesforcommittingreservedmemory.
#Anerrorreportfilewithmoreinformationissavedas:
#/usr/local/rocketmq/bin/hs_err_pid1997.log

解決方案：

是由於預設記憶體分配的太大了，超出了本機記憶體，直接OOM了。

修改bin/目錄下的如下兩個指令碼

runbroker.sh
runserver.sh

在這兩個腳本里都搜尋-server -Xms，將其記憶體分配小點，自己玩的話512MB就足夠了，夠夠的了！

4.3、啟動成功標識

namesrv啟動成功標識：

broker啟動成功標識：

二、RocketMQ控制檯的安裝

控制檯目前獲取方式有如下兩種：

第三方網站去下載現成的，比如csdn等。
官方原始碼包自己編譯而成，官方沒有現成的。

我們這裡當然採取官方方式。

1、官方檔案

github倉庫

https://github.com/apache/rocketmq-externals

中文指南

https://github.com/apache/rocketmq-externals/blob/master/rocketmq-console/doc/1_0_0/UserGuide_CN.md

2、下載原始碼

https://codeload.github.com/apache/rocketmq-externals/zip/master

3、修改配置（可選）

我們下載完解壓後的檔案目錄如下：

修改rocketmq-consolesrcmainresourcesapplication.properties檔案的server.port就歐了。預設8080。

4、編譯打包

進入rocketmq-console，然後用maven進行編譯打包

mvncleanpackage-DskipTests

打包完會在target下生成我們spring boot的jar程式，直接java -jar啟動完事。

5、啟動控制檯

將編譯打包好的springboot程式扔到伺服器上，執行如下命令進行啟動

java-jarrocketmq-console-ng-1.0.1.jar--rocketmq.config.namesrvAddr=127.0.0.1:9876

如果想後臺啟動就nohup &

訪問一下看看效果：

三、測試

rocketmq給我們提供了測試工具和測試類，可以在安裝完很方便的進行測試。

0、準備工作

rocketmq給我們提供的預設測試工具在bin目錄下，叫tools.sh。我們測試前需要配置這個指令碼，為他指定namesrv地址才可以，否則測試傳送/消費訊息的時候會出現如下錯誤connect to null failed：

22:49:02.470[main]DEBUGi.n.u.i.l.InternalLoggerFactory-UsingSLF4Jasthedefaultloggingframework
RocketMQLog:WARNNoappenderscouldbefoundforlogger(io.netty.util.internal.PlatformDependent0).
RocketMQLog:WARNPleaseinitializetheloggersystemproperly.
java.lang.IllegalStateException:org.apache.rocketmq.remoting.exception.RemotingConnectException:connecttonullfailed

配置如下：

vimtools.sh
#在exportJAVA_HOME上面新增如下這段程式碼
exportNAMESRV_ADDR=localhost:9876

1、傳送訊息

./tools.shorg.apache.rocketmq.example.quickstart.Producer

成功的話會看到嘩嘩嘩的日誌，因為這個類會傳送1000條訊息到TopicTest這個Topic下。

2、消費訊息

./tools.shorg.apache.rocketmq.example.quickstart.Consumer

成功的話會看到嘩嘩嘩的日誌，因為這個類會消費TopicTest下的全部訊息。剛傳送的1000條都會被消費掉。

3、控制檯

傳送成功後我們自然也能來到管控臺去看訊息和消費情況等等等資訊

四、架構圖以及角色

1、架構圖

2、角色

2.1、Broker

理解成RocketMQ本身
broker主要用於producer和consumer接收和傳送訊息
broker會定時向nameserver提交自己的資訊
是訊息中介軟體的訊息儲存、轉發伺服器
每個Broker節點，在啟動時，都會遍歷NameServer列表，與每個NameServer建立長連線，註冊自己的資訊，之後定時上報

2.2、Nameserver

理解成zookeeper的效果，只是他沒用zk，而是自己寫了個nameserver來替代zk
底層由netty實現，提供了路由管理、服務註冊、服務發現的功能，是一個無狀態節點
nameserver是服務發現者，叢集中各個角色（producer、broker、consumer等）都需要定時向nameserver上報自己的狀態，以便互相發現彼此，超時不上報的話，nameserver會把它從列表中剔除
nameserver可以部署多個，當多個nameserver存在的時候，其他角色同時向他們上報資訊，以保證高可用，
NameServer叢集間互不通訊，沒有主備的概念
nameserver記憶體式儲存，nameserver中的broker、topic等資訊預設不會持久化，所以他是無狀態節點

2.3、Producer

訊息的生產者
隨機選擇其中一個NameServer節點建立長連線，獲得Topic路由資訊（包括topic下的queue，這些queue分佈在哪些broker上等等）
接下來向提供topic服務的master建立長連線（因為rocketmq只有master才能寫訊息），且定時向master傳送心跳

2.4、Consumer

訊息的消費者
通過NameServer叢集獲得Topic的路由資訊，連線到對應的Broker上消費訊息
由於Master和Slave都可以讀取訊息，因此Consumer會與Master和Slave都建立連線進行消費訊息

3、核心流程

Broker都註冊到Nameserver上
Producer發訊息的時候會從Nameserver上獲取發訊息的topic資訊
Producer向提供服務的所有master建立長連線，且定時向master傳送心跳
Consumer通過NameServer叢集獲得Topic的路由資訊
Consumer會與所有的Master和所有的Slave都建立連線進行監聽新訊息

五、核心概念

1、Message

訊息載體。Message傳送或者消費的時候必須指定Topic。Message有一個可選的Tag項用於過濾訊息，還可以新增額外的鍵值對。

2、topic

訊息的邏輯分類，發訊息之前必須要指定一個topic才能發，就是將這條訊息傳送到這個topic上。消費訊息的時候指定這個topic進行消費。就是邏輯分類。

3、queue

1個Topic會被分為N個Queue，數量是可配置的。message本身其實是儲存到queue上的，消費者消費的也是queue上的訊息。多說一嘴，比如1個topic4個queue，有5個Consumer都在消費這個topic，那麼會有一個consumer浪費掉了，因為負載均衡策略，每個consumer消費1個queue，5>4，溢位1個，這個會不工作。

4、Tag

Tag 是 Topic 的進一步細分，顧名思義，標籤。每個傳送的時候訊息都能打tag，消費的時候可以根據tag進行過濾，選擇性消費。

5、Message Model

訊息模型：叢集（Clustering）和廣播（Broadcasting）

6、Message Order

訊息順序：順序（Orderly）和併發（Concurrently）

7、Producer Group

訊息生產者組

8、Consumer Group

訊息消費者組

六、ACK

首先要明確一點：ACK機制是發生在Consumer端的，不是在Producer端的。也就是說Consumer消費完訊息後要進行ACK確認，如果未確認則代表是消費失敗，這時候Broker會進行重試策略（僅叢集模式會重試）。ACK的意思就是：Consumer說：ok，我消費成功了。這條訊息給我標記成已消費吧。

七、消費模式

1、叢集模式（Clustering）

1.1、圖解

1.2、特點

每條訊息只需要被處理一次，broker只會把訊息傳送給消費叢集中的一個消費者
在訊息重投時，不能保證路由到同一臺機器上
消費狀態由broker維護

2、廣播模式（Broadcasting）

2.1、圖解

2.2、特點

消費進度由consumer維護
保證每個消費者都消費一次訊息
消費失敗的訊息不會重投

八、Java API

說明：

RocketMQ服務端版本為目前最新版：4.7.0
Java客戶端版本採取的目前最新版：4.7.0

pom如下

<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.7.0</version>
</dependency>

1、Producer

發訊息肯定要必備如下幾個條件：

指定生產組名（不能用預設的，會報錯）

配置namesrv地址（必須）

指定topic name（必須）

指定tag/key（可選）

驗證訊息是否傳送成功：訊息傳送完後可以啟動消費者進行消費，也可以去管控臺上看訊息是否存在。

1.1、send（同步）

publicclassProducer{
publicstaticvoidmain(String[]args)throwsException{
//指定生產組名為my-producer
DefaultMQProducerproducer=newDefaultMQProducer("my-producer");
//配置namesrv地址
producer.setNamesrvAddr("124.57.180.156:9876");
//啟動Producer
producer.start();
//建立訊息物件，topic為：myTopic001，訊息內容為：hello world
Messagemsg=newMessage("myTopic001","helloworld".getBytes());
//傳送訊息到mq，同步的
SendResultresult=producer.send(msg);
System.out.println("傳送訊息成功！result is :"+result);
//關閉Producer
producer.shutdown();
System.out.println("生產者 shutdown！");
}
}

輸出結果：

傳送訊息成功！result is : SendResult [sendStatus=SEND_OK, msgId=A9FE854140F418B4AAC26F7973910000, offsetMsgId=7B39B49D00002A9F00000000000589BE,messageQueue=MessageQueue[topic=myTopic001,brokerName=broker-a,queueId=0],queueOffset=7]
生產者 shutdown！

1.2、send（批量）

publicclassProducerMultiMsg{
publicstaticvoidmain(String[]args)throwsException{
//指定生產組名為my-producer
DefaultMQProducerproducer=newDefaultMQProducer("my-producer");
//配置namesrv地址
producer.setNamesrvAddr("124.57.180.156:9876");
//啟動Producer
producer.start();

Stringtopic="myTopic001";
//建立訊息物件，topic為：myTopic001，訊息內容為：hello world1/2/3
Messagemsg1=newMessage(topic,"helloworld1".getBytes());
Messagemsg2=newMessage(topic,"helloworld2".getBytes());
Messagemsg3=newMessage(topic,"helloworld3".getBytes());
//建立訊息物件的集合，用於批量傳送
List<Message>msgs=newArrayList<>();
msgs.add(msg1);
msgs.add(msg2);
msgs.add(msg3);
//批量傳送的api的也是send()，只是他的過載方法支援List<Message>，同樣是同步傳送。
SendResultresult=producer.send(msgs);
System.out.println("傳送訊息成功！result is :"+result);
//關閉Producer
producer.shutdown();
System.out.println("生產者 shutdown！");
}
}

輸出結果：

傳送訊息成功！result is : SendResult [sendStatus=SEND_OK, msgId=A9FE854139C418B4AAC26F7D13770000,A9FE854139C418B4AAC26F7D13770001,A9FE854139C418B4AAC26F7D13770002, offsetMsgId=7B39B49D00002A9F0000000000058A62,7B39B49D00002A9F0000000000058B07,7B39B49D00002A9F0000000000058BAC,messageQueue=MessageQueue[topic=myTopic001,brokerName=broker-a,queueId=0],queueOffset=8]
生產者 shutdown！

從結果中可以看到只有一個msgId，所以可以發現雖然是三條訊息物件，但是卻只發送了一次，大大節省了client與server的開銷。

錯誤情況：

批量傳送的topic必須是同一個，如果message物件指定不同的topic，那麼批量傳送的時候會報錯：

Exceptioninthread"main"org.apache.rocketmq.client.exception.MQClientException:FailedtoinitiatetheMessageBatch
Formoreinformation,pleasevisittheurl,http://rocketmq.apache.org/docs/faq/
atorg.apache.rocketmq.client.producer.DefaultMQProducer.batch(DefaultMQProducer.java:950)
atorg.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:898)
atcom.chentongwei.mq.rocketmq.ProducerMultiMsg.main(ProducerMultiMsg.java:29)
Causedby:java.lang.UnsupportedOperationException:Thetopicofthemessagesinonebatchshouldbethesame
atorg.apache.rocketmq.common.message.MessageBatch.generateFromList(MessageBatch.java:58)
atorg.apache.rocketmq.client.producer.DefaultMQProducer.batch(DefaultMQProducer.java:942)
...2more

1.3、sendCallBack（非同步）

publicclassProducerASync{
publicstaticvoidmain(String[]args)throwsException{
//指定生產組名為my-producer
DefaultMQProducerproducer=newDefaultMQProducer("my-producer");
//配置namesrv地址
producer.setNamesrvAddr("124.57.180.156:9876");
//啟動Producer
producer.start();

//建立訊息物件，topic為：myTopic001，訊息內容為：hello world async
Messagemsg=newMessage("myTopic001","helloworldasync".getBytes());
//進行非同步傳送，通過SendCallback介面來得知傳送的結果
producer.send(msg,newSendCallback(){
//傳送成功的回撥介面
@Override
publicvoidonSuccess(SendResultsendResult){
System.out.println("傳送訊息成功！result is :"+sendResult);
}
//傳送失敗的回撥介面
@Override
publicvoidonException(Throwablethrowable){
throwable.printStackTrace();
System.out.println("傳送訊息失敗！result is :"+throwable.getMessage());
}
});

producer.shutdown();
System.out.println("生產者 shutdown！");
}
}

輸出結果：

生產者 shutdown！
java.lang.IllegalStateException:org.apache.rocketmq.remoting.exception.RemotingConnectException:connectto[124.57.180.156:9876]failed
atorg.apache.rocketmq.client.impl.factory.MQClientInstance.updateTopicRouteInfoFromNameServer(MQClientInstance.java:681)
atorg.apache.rocketmq.client.impl.factory.MQClientInstance.updateTopicRouteInfoFromNameServer(MQClientInstance.java:511)
atorg.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.tryToFindTopicPublishInfo(DefaultMQProducerImpl.java:692)
atorg.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:556)
atorg.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.access$300(DefaultMQProducerImpl.java:97)
atorg.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl$4.run(DefaultMQProducerImpl.java:510)
atjava.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
atjava.util.concurrent.FutureTask.run(FutureTask.java:266)
atjava.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
atjava.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
atjava.lang.Thread.run(Thread.java:745)
Causedby:org.apache.rocketmq.remoting.exception.RemotingConnectException:connectto[124.57.180.156:9876]failed
atorg.apache.rocketmq.remoting.netty.NettyRemotingClient.getAndCreateNameserverChannel(NettyRemotingClient.java:441)
atorg.apache.rocketmq.remoting.netty.NettyRemotingClient.getAndCreateChannel(NettyRemotingClient.java:396)
atorg.apache.rocketmq.remoting.netty.NettyRemotingClient.invokeSync(NettyRemotingClient.java:365)
atorg.apache.rocketmq.client.impl.MQClientAPIImpl.getTopicRouteInfoFromNameServer(MQClientAPIImpl.java:1371)
atorg.apache.rocketmq.client.impl.MQClientAPIImpl.getTopicRouteInfoFromNameServer(MQClientAPIImpl.java:1361)
atorg.apache.rocketmq.client.impl.factory.MQClientInstance.updateTopicRouteInfoFromNameServer(MQClientInstance.java:624)
...10more
傳送訊息失敗！result is : org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [124.57.180.156:9876]failed

為啥報錯了？很簡單，他是非同步的，從結果就能看出來，由於是非同步的，我還沒傳送到mq呢，你就先給我shutdown了。肯定不行，所以我們在shutdown前面sleep 1s在看效果

publicclassProducerASync{
publicstaticvoidmain(String[]args)throwsException{
//指定生產組名為my-producer
DefaultMQProducerproducer=newDefaultMQProducer("my-producer");
//配置namesrv地址
producer.setNamesrvAddr("124.57.180.156:9876");
//啟動Producer
producer.start();

//建立訊息物件，topic為：myTopic001，訊息內容為：hello world async
Messagemsg=newMessage("myTopic001","helloworldasync".getBytes());
//進行非同步傳送，通過SendCallback介面來得知傳送的結果
producer.send(msg,newSendCallback(){
//傳送成功的回撥介面
@Override
publicvoidonSuccess(SendResultsendResult){
System.out.println("傳送訊息成功！result is :"+sendResult);
}
//傳送失敗的回撥介面
@Override
publicvoidonException(Throwablethrowable){
throwable.printStackTrace();
System.out.println("傳送訊息失敗！result is :"+throwable.getMessage());
}
});

Thread.sleep(1000);

producer.shutdown();
System.out.println("生產者 shutdown！");
}
}

輸出結果：

傳送訊息成功！result is : SendResult [sendStatus=SEND_OK, msgId=A9FE854106E418B4AAC26F8719B20000, offsetMsgId=7B39B49D00002A9F0000000000058CFC,messageQueue=MessageQueue[topic=myTopic001,brokerName=broker-a,queueId=1],queueOffset=2]
生產者 shutdown！

1.4、sendOneway

publicclassProducerOneWay{
publicstaticvoidmain(String[]args)throwsException{
//指定生產組名為my-producer
DefaultMQProducerproducer=newDefaultMQProducer("my-producer");
//配置namesrv地址
producer.setNamesrvAddr("124.57.180.156:9876");
//啟動Producer
producer.start();

//建立訊息物件，topic為：myTopic001，訊息內容為：hello world oneway
Messagemsg=newMessage("myTopic001","helloworldoneway".getBytes());
//效率最高，因為oneway不關心是否傳送成功，我就投遞一下我就不管了。所以返回是void
producer.sendOneway(msg);
System.out.println("投遞訊息成功！，注意這裡是投遞成功，而不是傳送訊息成功哦！因為我sendOneway也不知道到底成沒成功，我沒返回值的。");
producer.shutdown();
System.out.println("生產者 shutdown！");
}
}

輸出結果：

投遞訊息成功！，注意這裡是投遞成功，而不是傳送訊息成功哦！因為我sendOneway也不知道到底成沒成功，我沒返回值的。
生產者 shutdown！

1.5、效率對比

sendOneway > sendCallBack > send批量 > send單條

很容易理解，sendOneway不求結果，我就負責投遞，我不管你失敗還是成功，相當於中轉站，來了我就扔出去，我不進行任何其他處理。所以最快。

而sendCallBack是非同步傳送肯定比同步的效率高。

send批量和send單條的效率也是分情況的，如果只有1條msg要發，那還搞毛批量，直接send單條完事。

2、Consumer

每個consumer只能關注一個topic。

發訊息肯定要必備如下幾個條件：

指定消費組名（不能用預設的，會報錯）
配置namesrv地址（必須）
指定topic name（必須）
指定tag/key（可選）

2.1、CLUSTERING

叢集模式，預設。

比如啟動五個Consumer，Producer生產一條訊息後，Broker會選擇五個Consumer中的其中一個進行消費這條訊息，所以他屬於點對點消費模式。

publicclassConsumer{
publicstaticvoidmain(String[]args)throwsException{
//指定消費組名為my-consumer
DefaultMQPushConsumerconsumer=newDefaultMQPushConsumer("my-consumer");
//配置namesrv地址
consumer.setNamesrvAddr("124.57.180.156:9876");
//訂閱topic：myTopic001 下的全部訊息（因為是*，*指定的是tag標籤，代表全部訊息，不進行任何過濾）
consumer.subscribe("myTopic001","*");
//註冊監聽器，進行訊息訊息。
consumer.registerMessageListener(newMessageListenerConcurrently(){
@Override
publicConsumeConcurrentlyStatusconsumeMessage(List<MessageExt>msgs,ConsumeConcurrentlyContextconsumeConcurrentlyContext){
for(MessageExtmsg:msgs){
Stringstr=newString(msg.getBody());
//輸出訊息內容
System.out.println(str);
}
//預設情況下，這條訊息只會被一個consumer消費，這叫點對點消費模式。也就是叢集模式。
//ack確認
returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
//啟動消費者
consumer.start();
System.out.println("Consumerstart");
}
}

2.2、BROADCASTING

廣播模式。

比如啟動五個Consumer，Producer生產一條訊息後，Broker會把這條訊息廣播到五個Consumer中，這五個Consumer分別消費一次，每個都消費一次。

//程式碼裡只需要新增如下這句話即可：
consumer.setMessageModel(MessageModel.BROADCASTING);

2.3、兩種模式對比

叢集預設是預設的，廣播模式是需要手動配置。
一條訊息：叢集模式下的多個Consumer只會有一個Consumer消費。廣播模式下的每一個Consumer都會消費這條訊息。
廣播模式下，傳送一條訊息後，會被當前被廣播的所有Consumer消費，但是後面新加入的Consumer不會消費這條訊息，很好理解：村裡面大喇叭喊了全村來領雞蛋，第二天你們村新來個人，那個人肯定聽不到昨天大喇叭喊的訊息呀。

3、TAG&&KEY

傳送/消費訊息的時候可以指定tag/key來進行過濾訊息，支援萬用字元。*代表消費此topic下的全部訊息，不進行過濾。

看下org.apache.rocketmq.common.message.Message原始碼可以發現發訊息的時候可以指定tag和keys：

publicMessage(Stringtopic,Stringtags,Stringkeys,byte[]body){
this(topic,tags,keys,0,body,true);
}

比如：

publicclassProducerTagsKeys{
publicstaticvoidmain(String[]args)throwsException{
//指定生產組名為my-producer
DefaultMQProducerproducer=newDefaultMQProducer("my-producer");
//配置namesrv地址
producer.setNamesrvAddr("124.57.180.156:9876");
//啟動Producer
producer.start();
//建立訊息物件，topic為：myTopic001，訊息內容為：hello world，且tags為：test-tags，keys為test-keys
Messagemsg=newMessage("myTopic001","test-tags","test-keys","helloworld".getBytes());
//傳送訊息到mq，同步的
SendResultresult=producer.send(msg);
System.out.println("傳送訊息成功！result is :"+result);
//關閉Producer
producer.shutdown();
System.out.println("生產者 shutdown！");
}
}

輸出結果：

傳送訊息成功！result is : SendResult [sendStatus=SEND_OK, msgId=A9FE854149DC18B4AAC26FA4B7200000, offsetMsgId=7B39B49D00002A9F0000000000058DA6,messageQueue=MessageQueue[topic=myTopic001,brokerName=broker-a,queueId=3],queueOffset=3]
生產者 shutdown！

檢視管控臺，可以發現tags和keys已經生效了：

消費的時候如果指定*那就是此topic下的全部訊息，我們可以指定字首萬用字元，比如：

//這樣就只會消費myTopic001下的tag為test-*開頭的訊息。
consumer.subscribe("myTopic001","test-*");

//代表訂閱Topic為myTopic001下的tag為TagA或TagB的所有訊息
consumer.subscribe("myTopic001","TagA||TagB");

還支援SQL表示式過濾，不是很常用。不BB了。

4、常見錯誤

4.1、sendDefaultImpl call timeout

4.1.1、異常

Exceptioninthread"main"org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException:sendDefaultImplcalltimeout
atorg.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:666)
atorg.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1342)
atorg.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1288)
atorg.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:324)
atcom.chentongwei.mq.rocketmq.Producer.main(Producer.java:18)

4.1.2、解決

1.如果你是雲伺服器，首先檢查安全組是否允許9876這個埠訪問，是否開啟了防火牆，如果開啟了的話是否將9876映射了出去。

2.修改配置檔案broker.conf，加上：

brokerIP1=我用的是阿里雲伺服器，這裡是我的公網IP

啟動namesrv和broker的時候加上本機IP（我用的是阿里雲伺服器，這裡是我的公網IP）：

./bin/mqnamesrv-nIP:9876
./bin/mqbroker-nIP:9876-cconf/broker.conf

4.2、No route info of this topic

4.2.1、異常

Exceptioninthread"main"org.apache.rocketmq.client.exception.MQClientException:Norouteinfoofthistopic:myTopic001
Seehttp://rocketmq.apache.org/docs/faq/forfurtherdetails.
atorg.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:684)
atorg.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1342)
atorg.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1288)
atorg.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:324)
atcom.chentongwei.mq.rocketmq.Producer.main(Producer.java:18)

4.2.2、解決

很明顯傳送成功了，不再是剛才的超時了，但是告訴我們沒有這個topic。那不能每次都手動建立呀，所以啟動broker的時候可以指定引數讓broker為我們自動建立。如下

./bin/mqbroker-nIP:9876-cconf/broker.confautoCreateTopicEnable=true

RocketMQ入門到入土（一）新手也能看懂的原理和實戰！

一、RocketMQ的安裝

1、檔案

2、下載

3、準備工作

3.1、解壓

3.2、安裝jdk

4、啟動

4.1、啟動namesrv

4.2、啟動broker

4.3、啟動成功標識

二、RocketMQ控制檯的安裝

1、官方檔案

2、下載原始碼

3、修改配置（可選）

4、編譯打包

5、啟動控制檯

三、測試

0、準備工作

1、傳送訊息

2、消費訊息

3、控制檯

四、架構圖以及角色

1、架構圖

2、角色

2.1、Broker

2.2、Nameserver

2.3、Producer

2.4、Consumer

3、核心流程

五、核心概念

1、Message

2、topic

3、queue

4、Tag

5、Message Model

6、Message Order

7、Producer Group

8、Consumer Group

六、ACK

七、消費模式

1、叢集模式（Clustering）

1.1、圖解

1.2、特點

2、廣播模式（Broadcasting）

2.1、圖解

2.2、特點

八、Java API

1、Producer

1.1、send（同步）

1.2、send（批量）

1.3、sendCallBack（非同步）

1.4、sendOneway

1.5、效率對比

2、Consumer

2.1、CLUSTERING

2.2、BROADCASTING

2.3、兩種模式對比

3、TAG&&KEY

4、常見錯誤

4.1、sendDefaultImpl call timeout

4.1.1、異常

4.1.2、解決

4.2、No route info of this topic

4.2.1、異常

4.2.2、解決

相關推薦