一、安裝

環境：

Linux version 2.6.32-573.el6.x86_64
RocketMQ 4.1.0
java 1.8
maven 3.3.9 

步驟：

1.下載原始碼並編譯

git clone https://github.com/apache/incubator-rocketmq.git 
        mvn clean package install -Prelease-all assembly:assembly -U

在編譯過程中需要下載很多的maven依賴包，建議還是更改自己自己的maven setting.xml 為國內映象伺服器地址。
oschina mirror url：

2.啟動nameserver

在使用之前先簡單瞭解一下這個到底是什麼？

在分散式協作系統中，比較常用的是zookeeper，用來保證整個分散式系統的服務之間的選舉，資料的釋出和訂閱，分散式鎖的控制等。而rocketmq是沒有接入zookeeper，而是通過自己內部的nameserver來維護活躍的broker列表，這樣broker僅僅只維護資料的訊息轉發、儲存以及master slave之間的同步。

進入到編譯完成之後的target檔案，會存在一個apache-rocketmq-all資料夾，這是編譯完成之後的檔案也是需要得到的最終檔案。

• a)配置ROCKETMQ_HOME
  在啟動之前需要配置ROCKETMQ_HOME環境變數，如下

vim /etc/profile
export ROCKETMQ_HOME=/usr/local/rocketmq/incubator-rocketmq/target/apache-rocketmq-all
PATH=$JAVA_HOME/bin:$M2_HOME/bin:$ROCKETMQ_HOME/bin:$PATH
source /etc/profile

通過echo $ROCKETMQ_HOME ，可以檢視ROCKETMQ_HOME 是否配置完成。

• b)授權指定sh指令碼檔案許可權
  進入$ROCKETMQ_HOME/bin 資料夾 執行

chmod +x mqadmin mqbroker mqfiltersrv mqshutdown  mqnamesrv

• c)啟動nameserver

nohup sh bin/mqnamesrv &

啟動之後檢視位於bin同級目錄下的nohup.out檔案

tailf nohup.out

如圖表示啟動成功

3.啟動broker

• a)配置NAMESRV_ADDR

 vim /etc/profile
export NAMESRV_ADDR=192.168.2.233:9876

儲存

• b)nohup mqbroker >/var/log/mq.log &

如上圖表示broker啟動成功

至此，rocketmq 安裝完畢。

二、入門例項

producer

package test.mq;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;

/**
 * @Author chenfanglin 【[email protected]】
 * @Date 2017/4/2118:00
 */
public class TestProducerRocketMQ {

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

        DefaultMQProducer producer = new DefaultMQProducer("producerGroupName");
        producer.setNamesrvAddr("192.168.2.235:9876");
        producer.start();
        try {
            {
                //topic:主題 testTopic.tags:主題下的tag tagA.keys:keyA.body:具體業務訊息體
                Message msg = new Message("testTopic","tagA","keyA",("Hello World tagA!").getBytes());
                for(int i=100;i>0;i--){
                    if(i%2==0){
                        SendResult sendResult = producer.send(msg);
                        Thread.sleep(1000);
                        System.out.println("tagA send result："+sendResult);
                    }else{
                        msg = new Message("testTopic","tagB","keyB",("Hello World tagB!").getBytes());
                        SendResult sendResult = producer.send(msg);
                        Thread.sleep(1000);
                        System.out.println("tagB send result:"+sendResult);
                    }
                }
            }
        }
        catch (Exception e) {
            e.printStackTrace();
        }
        //登出producer
        Runtime.getRuntime().addShutdownHook(new Thread(() -> producer.shutdown()));
        System.exit(0);
    }
}

consumer

package test.mq;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @Author chenfanglin 【[email protected]】
 * @Date 2017/4/2118:04
 */
public class TestConsumerRocketMQ {
    public static void main(String[] args) throws InterruptedException, MQClientException {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("testProducerGroupName");
        consumer.setNamesrvAddr("192.168.2.235:9876");

        //訂閱testTopic主題下，標籤為tagA或者tagB的訊息
        consumer.subscribe("testTopic", "tagA || tagB");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
                MessageExt msg = msgs.get(0);
                if (msg.getTopic().equals("testTopic")) {
                    if (msg.getTags() != null && msg.getTags().equals("tagA")) {
                        // 執行tagA的消費
                        String message = new String(msg.getBody());
                        System.out.println("receive tagA message:"+message);
                    }
                    else if (msg.getTags() != null && msg.getTags().equals("tagB")) {
                        // 執行tagB的消費
                        String message = new String(msg.getBody());
                        System.out.println("receive tagB message:"+message);
                    }

                }
                //回執訊息確認
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("Consumer Started.");
    }
}

資料：