【linux】RocketMQ:一個純java的開源訊息中介軟體–開發測試環境搭建

【linux】RocketMQ:一個純java的開源訊息中介軟體–開發測試環境搭建

一、簡介

    RocketMQ的前身是Metaq,當 Metaq 3.0釋出時,產品名稱改為RocketMQ

    MetaQ2.x版本由於依賴了alibaba公司內部其他系統,對於公司外部使用者使用不夠友好,推薦使用3.0版本。

     專案地址:
https://github.com/alibaba/RocketMQ

二、安裝RocketMQ

     安裝RocketMQ需要jdk1.6, maven,git環境

     如果本機沒有安裝git,請使用如下命令安裝

     sudo apt-get install git

     具體安裝步驟可以參考
RocketMQ
專案組給出的步驟,參見:

     https://github.com/alibaba/RocketMQ/wiki/Quick-Start

git clone https://github.com/alibaba/RocketMQ.git
cd RocketMQ
sh install.sh
cd devenv


安裝完成後,因為install.sh指令碼中建立devenv 符號連結寫錯了目錄,需要在RocketMQ目錄下執行如下命令:  

       rm -rf devenv

       ln -s target/alibaba-rocketmq-3.0.7/alibaba-rocketmq devenv

     啟動RocketMQ

       cd devenv/bin

       nohup sh mqnamesrv &

       nohup sh mqbroker -n “192.168.1.110:9876” &        //因為我電腦在區域網中的ip是192.168.1.110,所以要這麼寫

       more nohup.out

       如果顯示:

       The Name Server boot success.
       The broker[vdata.kt, 192.168.1.110:10911] boot success.    //這個ip跟上面的那個ip應該在同一個地址段內

       則NameServer,Broker啟動成功

三、在eclipse中開發測試

       1.建立一個maven專案,其pom.xml的內容見pom.xml

       2. 編寫訊息產生者Producer,見Producer.java

       3.編寫訊息消費者Consumer,見PushConsumer.java

pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.catt</groupId>
<artifactId>RocketMQTest</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>RocketMQTest</name>
<url>http://maven.apache.org</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>com.alibaba.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>3.0.7</version>
</dependency>
<dependency>
<groupId>com.alibaba.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
<version>3.0.7</version>
<type>pom</type>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.1.1</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-core</artifactId>
<version>1.1.1</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.10</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>

Producer.java

預設情況下,一臺伺服器只能啟動一個Producer或Consumer例項,所以如果需要在一臺伺服器啟動多個例項,需要設定例項的名稱

producer.setNamesrvAddr(“192.168.230.128:9876”);
producer.setInstanceName(“Producer”);

package com.catt.rocketmq.example;
import java.util.concurrent.TimeUnit;
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;
public class Producer {
public static void main(String[] args) throws MQClientException,
InterruptedException {
/**
* 一個應用建立一個Producer,由應用來維護此物件,可以設定為全域性物件或者單例<br>
* 注意:ProducerGroupName需要由應用來保證唯一<br>
* ProducerGroup這個概念傳送普通的訊息時,作用不大,但是傳送分散式事務訊息時,比較關鍵,
* 因為伺服器會回查這個Group下的任意一個Producer
*/
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
producer.setNamesrvAddr("192.168.230.128:9876");
producer.setInstanceName("Producer");
/**
* Producer物件在使用之前必須要呼叫start初始化,初始化一次即可<br>
* 注意:切記不可以在每次傳送訊息時,都呼叫start方法
*/
producer.start();
/**
* 下面這段程式碼表明一個Producer物件可以傳送多個topic,多個tag的訊息。
* 注意:send方法是同步呼叫,只要不拋異常就標識成功。但是傳送成功也可會有多種狀態,<br>
* 例如訊息寫入Master成功,但是Slave不成功,這種情況訊息屬於成功,但是對於個別應用如果對訊息可靠性要求極高,<br>
* 需要對這種情況做處理。另外,訊息可能會存在傳送失敗的情況,失敗重試由應用來處理。
*/
for (int i = 0; i < 100; i  ) {
try {
{
Message msg = new Message("TopicTest1",// topic
"TagA",// tag
"OrderID001",// key
("Hello MetaQ").getBytes());// body
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
{
Message msg = new Message("TopicTest2",// topic
"TagB",// tag
"OrderID0034",// key
("Hello MetaQ").getBytes());// body
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
{
Message msg = new Message("TopicTest3",// topic
"TagC",// tag
"OrderID061",// key
("Hello MetaQ").getBytes());// body
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
} catch (Exception e) {
e.printStackTrace();
}
TimeUnit.MILLISECONDS.sleep(1000);
}
/**
* 應用退出時,要呼叫shutdown來清理資源,關閉網路連線,從MetaQ伺服器上登出自己
* 注意:我們建議應用在JBOSS、Tomcat等容器的退出鉤子裡呼叫shutdown方法
*/
producer.shutdown();
}
}

PushConsumer.java

package com.catt.rocketmq.example;
import java.util.List;
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.message.MessageExt;
public class PushConsumer {
/**
* 當前例子是PushConsumer用法,使用方式給使用者感覺是訊息從RocketMQ伺服器推到了應用客戶端。<br>
* 但是實際PushConsumer內部是使用長輪詢Pull方式從MetaQ伺服器拉訊息,然後再回撥使用者Listener方法<br>
*/
public static void main(String[] args) throws InterruptedException,
MQClientException {
/**
* 一個應用建立一個Consumer,由應用來維護此物件,可以設定為全域性物件或者單例<br>
* 注意:ConsumerGroupName需要由應用來保證唯一
*/
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(
"ConsumerGroupName");
consumer.setNamesrvAddr("192.168.230.128:9876");
consumer.setInstanceName("Consumber");
/**
* 訂閱指定topic下tags分別等於TagA或TagC或TagD
*/
consumer.subscribe("TopicTest1", "TagA || TagC || TagD");
/**
* 訂閱指定topic下所有訊息<br>
* 注意:一個consumer物件可以訂閱多個topic
*/
consumer.subscribe("TopicTest2", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
/**
* 預設msgs裡只有一條訊息,可以通過設定consumeMessageBatchMaxSize引數來批量接收訊息
*/
@Override
public ConsumeConcurrentlyStatus consumeMessage(
List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
System.out.println(Thread.currentThread().getName()
" Receive New Messages: "   msgs.size());
MessageExt msg = msgs.get(0);
if (msg.getTopic().equals("TopicTest1")) {
// 執行TopicTest1的消費邏輯
if (msg.getTags() != null && msg.getTags().equals("TagA")) {
// 執行TagA的消費
System.out.println(new String(msg.getBody()));
} else if (msg.getTags() != null
&& msg.getTags().equals("TagC")) {
// 執行TagC的消費
} else if (msg.getTags() != null
&& msg.getTags().equals("TagD")) {
// 執行TagD的消費
}
} else if (msg.getTopic().equals("TopicTest2")) {
System.out.println(new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
/**
* Consumer物件在使用之前必須要呼叫start初始化,初始化一次即可<br>
*/
consumer.start();
System.out.println("Consumer Started.");
}
}

由於是非同步,所以執行不分先後。

在執行之前,先匯入https://github.com/alibaba/RocketMQ/releases這個頁面下的 alibaba-rocketmq-client-java-3.2.6.tar.gz 這個裡面所有的jar包。

如果執行過程中提示:缺少slf4j nop之類的,就下載個slf4j-nop-1.6.1.jar,匯入進去就可以了。

完整的是這樣:

原帖地址:http://www.tuicool.com/articles/mEBVneB