RocketMQ 叢集部署模式

• 單 master 模式

也就是隻有一個 master 節點，稱不上是叢集，一旦這個 master 節點宕機，那麼整個服務就不可用。

優點：部署簡單。
缺點：存在單點故障。
注意：該模式一般只用來個人學習，或者作為開發環境使用，生產環境不推薦使用該模式。

• 多 master 模式

多個 master 節點組成叢集，單個 master 節點宕機或者重啟對應用來說沒有影響。

優點：所有模式中效能最高。
缺點：單個 master 節點宕機期間，未被消費的訊息在節點恢復之前不可用，訊息的實時性會受到影響。
注意：使用同步刷盤可以保證訊息不丟失，同時每個 Topic 應該均勻分佈在叢集中每個節點，而不是隻在某個節點上，否則，該節點宕機就會對訂閱該 topic 的應用造成影響。

• 多 master 多 slave 非同步複製模式

在多 master 模式的基礎上，每個 master 節點都有至少一個對應的 slave。master 節點可讀可寫，但是 slave 只能讀不能寫，類似於 mysql 的主備模式。

優點： 在 master 宕機時，消費者可以從 slave 讀取訊息，訊息的實時性不會受影響，效能幾乎和多 master 一樣。
缺點：使用非同步複製的同步方式有可能會有訊息丟失的問題。

• 多 master 多 slave 同步雙寫模式

同多 master 多 slave 非同步複製模式類似，區別在於 master 和 slave 之間的資料同步方式是同步的。

優點：同步雙寫模式能保證資料不丟失。
缺點

RocketMQ 單主部署

鑑於是快速入門，我選擇的是第一種單 master 的部署模式。先說明一下我的安裝環境：

1. Centos 7.2
2. jdk 1.8
3. Maven 3.2.x
4. Git

這裡 git 可用可不用，主要是用來直接下載 github 上的原始碼。也可以選擇自己到
github 上下載，然後上傳到伺服器上。這裡以 git 操作為示例。

1. clone 原始碼並用 maven 編譯

> git clone https://github.com/alibaba/RocketMQ.git /opt/RocketMQ
> cd /opt/RocketMQ && mvn -Dmaven.test.skip=true clean package install assembly:assembly -U
> cd target/alibaba-rocketmq-broker/alibaba-rocketmq

2. 啟動 Name Server

> nohup sh /opt/RocketMQ/bin/mqnamesrv &
//執行 jps 檢視程序
> jps
25913 NamesrvStartup
//檢視日誌確保服務已正常啟動
> tail -f ~/logs/rocketmqlogs/namesrv.log
The Name Server boot success...

3. 啟動 broker

> nohup sh /opt/RocketMQ/bin/mqbroker -n localhost:9876 &
//執行 jps 檢視程序
> jps
25954 BrokerStartup
//檢視日誌確保服務已正常啟動
> tail -f ~/logs/rocketmqlogs/broker.log
The broker[broker-a, 10.1.54.121:10911] boot success...

4. 傳送和接收訊息

傳送/接收訊息之前，我們需要告訴客戶端 NameServer 地址。RocketMQ 提供了多種方式來實現這一目標。為簡單起見，我們使用環境變數 NAMESRV_ADDR。

> export NAMESRV_ADDR=localhost:9876
> sh /opt/RocketMQ/bin/tools.sh com.alibaba.rocketmq.example.quickstart.Producer
SendResult [sendStatus=SEND_OK, msgId= ...
> sh /opt/RocketMQ/bin/tools.sh com.alibaba.rocketmq.example.quickstart.Consumer
ConsumeMessageThread_%d Receive New Messages: [MessageExt...

5. 關閉服務

> sh /opt/RocketMQ/bin/mqshutdown broker
The mqbroker(36695) is running...
Send shutdown request to mqbroker(36695) OK
> sh /opt/RocketMQ/bin/mqshutdown namesrv
The mqnamesrv(36664) is running...
Send shutdown request to mqnamesrv(36664) OK

此處可能遇到的問題：

fatal: unable to access 'https://github.com/alibaba/RocketMQ.git/': Could not resolve host: github.com; Unknown error

解決辦法：一般是由於網路原因造成的，執行以下命令：

> ping github.com

確定可以 ping 通之後，再重新執行 git clone 命令。

2. 執行"mvn -Dmaven.test.skip=true clean package install assembly:assembly -U"編譯時，可能出現下載相關 jar 包很慢的情況。

這也是由於預設 maven 中央倉庫在國外的原因，可以根據需要在 /home/maven/conf/setting.xml 中的 <mirrors></mirrors> 新增以下內容後重新編譯：

<mirror>
    <id>aliyun</id>
    <mirrorOf>central</mirrorOf>
    <name>aliyun maven</name>
    <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
</mirror>

示例程式碼

• 生產者

public class Producer {
    public static void main(String[] args) throws MQClientException, InterruptedException {

        //宣告並初始化一個producer
        //需要一個producer group名字作為構造方法的引數，這裡為producer1
        DefaultMQProducer producer = new DefaultMQProducer("producer1");

        //設定NameServer地址,此處應改為實際NameServer地址，多個地址之間用；分隔
        //NameServer的地址必須有，但是也可以通過環境變數的方式設定，不一定非得寫死在程式碼裡
        producer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");

        //呼叫start()方法啟動一個producer例項
        producer.start();

        //傳送10條訊息到Topic為TopicTest，tag為TagA，訊息內容為“Hello RocketMQ”拼接上i的值
        for (int i = 0; i < 10; i++) {
            try {
                Message msg = new Message("TopicTest",// topic
                        "TagA",// tag
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)// body
                );

                //呼叫producer的send()方法傳送訊息
                //這裡呼叫的是同步的方式，所以會有返回結果
                SendResult sendResult = producer.send(msg);

                //列印返回結果，可以看到訊息傳送的狀態以及一些相關資訊
                System.out.println(sendResult);
            } catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }

        //傳送完訊息之後，呼叫shutdown()方法關閉producer
        producer.shutdown();
    }
}

• 消費者

public class Consumer {

    public static void main(String[] args) throws InterruptedException, MQClientException {

        //宣告並初始化一個consumer
        //需要一個consumer group名字作為構造方法的引數，這裡為consumer1
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer1");

        //同樣也要設定NameServer地址
        consumer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");

        //這裡設定的是一個consumer的消費策略
        //CONSUME_FROM_LAST_OFFSET 預設策略，從該佇列最尾開始消費，即跳過歷史訊息
        //CONSUME_FROM_FIRST_OFFSET 從佇列最開始開始消費，即歷史訊息（還儲存在broker的）全部消費一遍
        //CONSUME_FROM_TIMESTAMP 從某個時間點開始消費，和setConsumeTimestamp()配合使用，預設是半個小時以前
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        //設定consumer所訂閱的Topic和Tag，*代表全部的Tag
        consumer.subscribe("TopicTest", "*");

        //設定一個Listener，主要進行訊息的邏輯處理
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {

                System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);

                //返回消費狀態
                //CONSUME_SUCCESS 消費成功
                //RECONSUME_LATER 消費失敗，需要稍後重新消費
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //呼叫start()方法啟動consumer
        consumer.start();

        System.out.println("Consumer Started.");
    }
}