前言

本文將帶大家學習使用前端開發神器-charles，從基本的下載安裝到常見配置使用，為大家一一講解。

一、花式誇獎Charles

• 截取 Http 和 Https 網絡封包。
• 支持重髮網絡請求，方便後端調試。
• 支持修改網絡請求參數。
• 支持網絡請求的截獲並動態修改。
• 支持模擬慢速網絡。

好，騎上我心愛的小摩托，準備上路…

二、下載與安裝

官網下載傳送門

本文所使用的的版本為 mac V4.5.6版本，不同版本間的具體化差異，大家可留言交流。

Charles破解工具可通過關注公眾號「胡哥有話說」，回復關鍵字charles獲得。

三、簡單入門-抓包所有請求

1. 打開Charles，勾選Proxy下的macOS Proxy（如果是windows，此處為Windows Proxy）
2. 點擊Proxy->Start Recording，打開瀏覽器訪問任意頁面，可以在Charles中看到請求了。

很好，現在已經上路了，學習的路上永不堵車…

四、設置過濾請求

通過上面的操作，我們已經抓包了所有的請求，實際開發中可能是專門針對某些接口（如百度域名下的接口），我們可以專門配置過濾接口。

1. 臨時性過濾配置

   在展示界面的Filter中可進行條件過濾

   同時可在右側的settings中配置使用正則來進行過濾

2. 永久性過濾配置

   通過Proxy->Recording Settings->include中配置過濾條件

Ok，我們又前進了一大步

五、代理轉發請求

通過Charles的Map Remote和Map Local我們也可以配置代理轉發請求。

Map Remote

Map Remote 遠程映射，是將指定的網絡請求重定向到另一個網址

業務場景：
某些服務端的文件請求時限制某些特定域名(*.baidu.com)，我們使用localhost啟動項目時,會導致沒有權限訪問。通過配置Map Remote遠程映射解決問題。

配置路徑：
設置Tools->Map Remote

如圖上的配置，本地啟動的項目地址為：http://localhost:8080（或 http://127.0.0.1:8080），現在再訪問，可以使用路徑 http://test.baidu.com訪問即可。

注意Enable Map Remote一定要勾選，以及相應規則也要勾選，否則不會生效

Map Local

Map Local 本地映射，是指將指定的網絡請求重定向到本地的文件

業務場景：
在本地化的開發中，接口數據Mock；或者是線上環境問題排查時，將請求重定向到本地文件以方便排查。

配置路徑：
設置Tools->Map Local

通過如上圖的配置，請求 aa.baidu.com:443/index時，會被映射到本地 /xx/index.json

注意Enable Map Local一定要勾選，以及相關規則也要勾選，否則不會生效。

六、手機抓包

手機抓包請求也是我們日常開發中需要用到的，那如何利用Charles抓包手機請求呢。

1. 設置Charles的代理端口號
   通過設置Proxy->Proxy Settings->Proxies->HTTP Proxy下的Port端口號

2. 查看本地IP地址
   通過Charles的Help->Local IP Address查看，本機IP為xx.xx.xx.xx

3. 手機和電腦需要處於同於wifi網絡內

4. 手機wifi網絡配置

   以華為mate 30為例，選擇對應的wifi，選擇显示高級選項，設置代理為手動。
   設置服務器主機名為：xx.xx.xx.xx（剛才查看的電腦IP）
   設置服務器端口為：8888（剛才設置的port）
   點擊保存后，手機的請求就可以在Charles中查看啦…

   注意鏈接時，Charles會彈出授權窗口，要選擇Allow

七、限速設置

通過設置Proxy->Throttle Settings來進行速度限制

注意：一定要勾選 Enable Throttling選項

小結

以上是給大家分享的Charles的常見使用配置，如有相關問題可留言交流。

後記

以上就是胡哥今天給大家分享的內容，喜歡的小夥伴記得點贊、收藏呦，關注胡哥有話說，學習前端不迷路，歡迎多多留言交流…

胡哥有話說，專註於大前端技術領域，分享前端系統架構，框架實現原理，最新最高效的技術實踐！

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理
【其他文章推薦】

※USB CONNECTOR掌控什麼技術要點? 帶您認識其相關發展及效能

※台北網頁設計公司這麼多該如何選擇?

※智慧手機時代的來臨，RWD網頁設計為架站首選

※評比南投搬家公司費用收費行情懶人包大公開

※幫你省時又省力,新北清潔一流服務好口碑

※回頭車貨運收費標準

學任何技術都是兩步驟：

1. 搭建環境

2. helloworld

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

一、RocketMQ的安裝

1、文檔

官方網站

http://rocketmq.apache.org

GitHub

https://github.com/apache/rocketmq

2、下載

wget https://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、解壓

unzip rocketmq-all-4.7.0-bin-release.zip

3.2、安裝jdk

sudo yum install java-1.8.0-openjdk-devel

4、啟動

4.1、啟動namesrv

cd rocketmq-all-4.7.0-bin-release/bin
./mqnamesrv

4.2、啟動broker

cd rocketmq-all-4.7.0-bin-release/bin
./mqbroker -n localhost:9876

常見錯誤以及解決方案：

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

[root@node-113b bin]# ./mqbroker -n localhost:9876
Java HotSpot(TM) 64-Bit Server VM warning: INFO: os::commit_memory(0x00000005c0000000, 8589934592, 0) failed
; error='Cannot allocate memory' (errno=12)#
# There is insufficient memory for the Java Runtime Environment to continue.
# Native memory allocation (mmap) failed to map 8589934592 bytes for committing reserved memory.
# An error report file with more information is saved as:
# /usr/local/rocketmq/bin/hs_err_pid1997.log

解決方案：

是由於默認內存分配的太大了，超出了本機內存，直接OOM了。

修改bin/目錄下的如下兩個腳本

runbroker.sh
runserver.sh

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

4.3、啟動成功標識

namesrv啟動成功標識：

broker啟動成功標識：

二、RocketMQ控制台的安裝

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

1. 第三方網站去下載現成的，比如csdn等。

2. 官方源碼包自己編譯而成，官方沒有現成的。

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

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進行編譯打包

mvn clean package -DskipTests

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

5、啟動控制台

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

java -jar rocketmq-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] DEBUG i.n.u.i.l.InternalLoggerFactory - Using SLF4J as the default logging framework
RocketMQLog:WARN No appenders could be found for logger (io.netty.util.internal.PlatformDependent0).
RocketMQLog:WARN Please initialize the logger system properly.
java.lang.IllegalStateException: org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to null failed

配置如下：

vim tools.sh
# 在export JAVA_HOME上面添加如下這段代碼
export NAMESRV_ADDR=localhost:9876

1、發送消息

./tools.sh org.apache.rocketmq.example.quickstart.Producer

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

2、消費消息

./tools.sh org.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（同步）

public class Producer {
    public static void main(String[] args) throws Exception {
        // 指定生產組名為my-producer
        DefaultMQProducer producer = new DefaultMQProducer("my-producer");
        // 配置namesrv地址
        producer.setNamesrvAddr("124.57.180.156:9876");
        // 啟動Producer
        producer.start();
        // 創建消息對象，topic為：myTopic001，消息內容為：hello world
        Message msg = new Message("myTopic001", "hello world".getBytes());
        // 發送消息到mq，同步的
        SendResult result = 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（批量）

public class ProducerMultiMsg {
    public static void main(String[] args) throws Exception {
        // 指定生產組名為my-producer
        DefaultMQProducer producer = new DefaultMQProducer("my-producer");
        // 配置namesrv地址
        producer.setNamesrvAddr("124.57.180.156:9876");
        // 啟動Producer
        producer.start();

        String topic = "myTopic001";
        // 創建消息對象，topic為：myTopic001，消息內容為：hello world1/2/3
        Message msg1 = new Message(topic, "hello world1".getBytes());
        Message msg2 = new Message(topic, "hello world2".getBytes());
        Message msg3 = new Message(topic, "hello world3".getBytes());
        // 創建消息對象的集合，用於批量發送
        List<Message> msgs = new ArrayList<>();
        msgs.add(msg1);
        msgs.add(msg2);
        msgs.add(msg3);
        // 批量發送的api的也是send()，只是他的重載方法支持List<Message>，同樣是同步發送。
        SendResult result = 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，那麼批量發送的時候會報錯：

Exception in thread "main" org.apache.rocketmq.client.exception.MQClientException: Failed to initiate the MessageBatch
For more information, please visit the url, http://rocketmq.apache.org/docs/faq/
    at org.apache.rocketmq.client.producer.DefaultMQProducer.batch(DefaultMQProducer.java:950)
    at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:898)
    at com.chentongwei.mq.rocketmq.ProducerMultiMsg.main(ProducerMultiMsg.java:29)
Caused by: java.lang.UnsupportedOperationException: The topic of the messages in one batch should be the same
    at org.apache.rocketmq.common.message.MessageBatch.generateFromList(MessageBatch.java:58)
    at org.apache.rocketmq.client.producer.DefaultMQProducer.batch(DefaultMQProducer.java:942)
    ... 2 more

1.3、sendCallBack（異步）

public class ProducerASync {
    public static void main(String[] args) throws Exception {
       // 指定生產組名為my-producer
        DefaultMQProducer producer = new DefaultMQProducer("my-producer");
        // 配置namesrv地址
        producer.setNamesrvAddr("124.57.180.156:9876");
        // 啟動Producer
        producer.start();

        // 創建消息對象，topic為：myTopic001，消息內容為：hello world async
        Message msg = new Message("myTopic001", "hello world async".getBytes());
        // 進行異步發送，通過SendCallback接口來得知發送的結果
        producer.send(msg, new SendCallback() {
            // 發送成功的回調接口
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("發送消息成功！result is : " + sendResult);
            }
            // 發送失敗的回調接口
            @Override
            public void onException(Throwable throwable) {
                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: connect to [124.57.180.156:9876] failed
    at org.apache.rocketmq.client.impl.factory.MQClientInstance.updateTopicRouteInfoFromNameServer(MQClientInstance.java:681)
    at org.apache.rocketmq.client.impl.factory.MQClientInstance.updateTopicRouteInfoFromNameServer(MQClientInstance.java:511)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.tryToFindTopicPublishInfo(DefaultMQProducerImpl.java:692)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:556)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.access$300(DefaultMQProducerImpl.java:97)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl$4.run(DefaultMQProducerImpl.java:510)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [124.57.180.156:9876] failed
    at org.apache.rocketmq.remoting.netty.NettyRemotingClient.getAndCreateNameserverChannel(NettyRemotingClient.java:441)
    at org.apache.rocketmq.remoting.netty.NettyRemotingClient.getAndCreateChannel(NettyRemotingClient.java:396)
    at org.apache.rocketmq.remoting.netty.NettyRemotingClient.invokeSync(NettyRemotingClient.java:365)
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.getTopicRouteInfoFromNameServer(MQClientAPIImpl.java:1371)
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.getTopicRouteInfoFromNameServer(MQClientAPIImpl.java:1361)
    at org.apache.rocketmq.client.impl.factory.MQClientInstance.updateTopicRouteInfoFromNameServer(MQClientInstance.java:624)
    ... 10 more
發送消息失敗！result is : org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [124.57.180.156:9876] failed

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

public class ProducerASync {
    public static void main(String[] args) throws Exception {
       // 指定生產組名為my-producer
        DefaultMQProducer producer = new DefaultMQProducer("my-producer");
        // 配置namesrv地址
        producer.setNamesrvAddr("124.57.180.156:9876");
        // 啟動Producer
        producer.start();

        // 創建消息對象，topic為：myTopic001，消息內容為：hello world async
        Message msg = new Message("myTopic001", "hello world async".getBytes());
        // 進行異步發送，通過SendCallback接口來得知發送的結果
        producer.send(msg, new SendCallback() {
            // 發送成功的回調接口
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("發送消息成功！result is : " + sendResult);
            }
            // 發送失敗的回調接口
            @Override
            public void onException(Throwable throwable) {
                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

public class ProducerOneWay {
    public static void main(String[] args) throws Exception {
        // 指定生產組名為my-producer
        DefaultMQProducer producer = new DefaultMQProducer("my-producer");
        // 配置namesrv地址
        producer.setNamesrvAddr("124.57.180.156:9876");
        // 啟動Producer
        producer.start();

        // 創建消息對象，topic為：myTopic001，消息內容為：hello world oneway
        Message msg = new Message("myTopic001", "hello world oneway".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中的其中一個進行消費這條消息，所以他屬於點對點消費模式。

public class Consumer {
    public static void main(String[] args) throws Exception {
        // 指定消費組名為my-consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my-consumer");
        // 配置namesrv地址
        consumer.setNamesrvAddr("124.57.180.156:9876");
        // 訂閱topic：myTopic001 下的全部消息（因為是*，*指定的是tag標籤，代表全部消息，不進行任何過濾）
        consumer.subscribe("myTopic001", "*");
        // 註冊監聽器，進行消息消息。
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg : msgs) {
                    String str = new String(msg.getBody());
                    // 輸出消息內容
                    System.out.println(str);
                }
                // 默認情況下，這條消息只會被一個consumer消費，這叫點對點消費模式。也就是集群模式。
                // ack確認
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 啟動消費者
        consumer.start();
        System.out.println("Consumer start");
    }
}

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：

public Message(String topic, String tags, String keys, byte[] body) {
    this(topic, tags, keys, 0, body, true);
}

比如：

public class ProducerTagsKeys {
    public static void main(String[] args) throws Exception {
        // 指定生產組名為my-producer
        DefaultMQProducer producer = new DefaultMQProducer("my-producer");
        // 配置namesrv地址
        producer.setNamesrvAddr("124.57.180.156:9876");
        // 啟動Producer
        producer.start();
        // 創建消息對象，topic為：myTopic001，消息內容為：hello world，且tags為：test-tags，keys為test-keys
        Message msg = new Message("myTopic001", "test-tags", "test-keys", "hello world".getBytes());
        // 發送消息到mq，同步的
        SendResult result = 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、異常

Exception in thread "main" org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException: sendDefaultImpl call timeout
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:666)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1342)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1288)
    at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:324)
    at com.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 -n IP:9876
./bin/mqbroker -n IP:9876 -c conf/broker.conf

4.2、No route info of this topic

4.2.1、異常

Exception in thread "main" org.apache.rocketmq.client.exception.MQClientException: No route info of this topic: myTopic001
See http://rocketmq.apache.org/docs/faq/ for further details.
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:684)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1342)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1288)
    at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:324)
    at com.chentongwei.mq.rocketmq.Producer.main(Producer.java:18)

4.2.2、解決

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

./bin/mqbroker -n IP:9876 -c conf/broker.conf autoCreateTopicEnable=true

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理
【其他文章推薦】

※USB CONNECTOR掌控什麼技術要點? 帶您認識其相關發展及效能

※台北網頁設計公司這麼多該如何選擇?

※智慧手機時代的來臨，RWD網頁設計為架站首選

※評比南投搬家公司費用收費行情懶人包大公開

※幫你省時又省力,新北清潔一流服務好口碑

※回頭車貨運收費標準