RocketMQ：一個純java的開源訊息中介軟體--開發測試環境搭建
阿新 • • 發佈：2019-02-04
一、簡介
RocketMQ的前身是Metaq，當 Metaq 3.0釋出時，產品名稱改為 RocketMQ
MetaQ2.x版本由於依賴了alibaba公司內部其他系統，對於公司外部使用者使用不夠友好，推薦使用3.0版本。
二、安裝RocketMQ
安裝RocketMQ需要jdk1.6, maven,git環境
如果本機沒有安裝git，請使用如下命令安裝
yum 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 devenvrm -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.230.128:9876" &
more nohup.out
如果顯示：
The Name Server boot success.
The broker[vdata.kt, 192.168.230.128:10911] boot success.
則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.");
}
}