淘宝消息中间件RocketMQ的安装和简单使用
RocketMQ是什么？
RocketMQ是一款分布式、队列模型的消息中间件，具有以下特点：
能够保证严格的消息顺序
提供丰富的消息拉取模式
高效的订阅者水平扩展能力
实时的消息订阅机制
亿级消息堆积能力
RocketMQ是alibaba开源的java消息中间件。在github上开源，项目同时也在oschina上。地址：https://git.oschina.net/vintagewang/RocketMQ/
下载RocketMQ3.0.8 最新的是3.1.0 试了好多次，编译怎么都不行啊。
果断使用了3.0.8
https://github.com/alibaba/RocketMQ/archive/v3.0.8.zip 本人机器有点问题git居然安装不上（纳闷啊）直接下的zip包。
解压开来，找到项目中pom.xml并且把其中的
<parent>
<groupId>com.taobao</groupId>
<artifactId>parent</artifactId>
<version>1.0.2</version>
</parent>
给注释掉，不然弄死不能编译的，我可是折腾了好久好久啊，这是由于阿里开源不彻底导致的。
下面开始编译吧。cmd控制台进到解压文件夹的pom.xml目录 执行
mvn -Dmaven.test.skip=true clean package install assembly:assembly -U 或者直接运行根目录下的install.cmd 推荐控制台运行，能直观的看到错误信息。install.cmd出错了控制台就没了。注释了parent之后应该就不报错了。等待它去网上下载完的jar包，并且编译完吧。
编译完成之后，根目录下回多出来一个叫target的目录，将里面的alibaba-rocketmq-3.0.8文件夹拷贝到你的安装盘符下吧，
命令行进入到${alibaba-rocketmq-3.0.8}/bin 输入start mqnamesrv.exe 会弹出一个信息窗口，记录的是日志也可以写到文件中 后面接着写 >${你的日志存放目录} ，在日志文件中看到The name Server boot success 说明启动成功了，输入jps -v 检测
2636 -Djava.ext.dirs=F:\taobao\alibaba-rocketmq-3.0.8\alibaba-rocketmq\bin/../lib-Drocketmq.home.dir=F:\taobao\alibaba-rocketmq-3.0.8\alibaba-rocketmq\bin/..-XX:MaxNewSize=512M -XX:MaxPermSize=128M -XX:NewSize=256M -XX:PermSize=128M -Xms512m -Xmx1g exit abort
启动borker start mqbroker.exe -n 127.0.0.1:9876 同样的弹出一个窗口 看到success表示成功了，文件日志同上。
Java操作列子，来自开源项目中。
建立一个普通的maven项目
Pom中加入如下配置
<dependencies>
<dependency>
<groupId>com.alibaba.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>3.0.8</version>
</dependency>
<dependency>
<groupId>com.alibaba.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
<version>3.0.8</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>
消息生产者
package rocketmq_test.test;
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
*/
final DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.setInstanceName("Producer");
/**
* Producer对象在使用之前必须要调用start初始化，初始化一次即可<br>
* 注意：切记不可以在每次发送消息时，都调用start方法
*/
producer.start();
/**
* 下面这段代码表明一个Producer对象可以发送多个topic，多个tag的消息。
* 注意：send方法是同步调用，只要不抛异常就标识成功。但是发送成功也可会有多种状态，<br>
* 例如消息写入Master成功，但是Slave不成功，这种情况消息属于成功，但是对于个别应用如果对消息可靠性要求极高，<br>
* 需要对这种情况做处理。另外，消息可能会存在发送失败的情况，失败重试由应用来处理。
*/
for (int i = 0; i < 10; i++){
try {
{
Message msg = new Message("TopicTest1",// topic
"TagA",// tag
"OrderID001",// key
("Hello MetaQA").getBytes());// body
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
{
Message msg = new Message("TopicTest2",// topic
"TagB",// tag
"OrderID0034",// key
("Hello MetaQB").getBytes());// body
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
{
Message msg = new Message("TopicTest3",// topic
"TagC",// tag
"OrderID061",// key
("Hello MetaQC").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();
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
public void run() {
producer.shutdown();
}
}));
System.exit(0);
}
}
消费者处理消息
package rocketmq_test.test;
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 Consumer {
/**
* 当前例子是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("127.0.0.1: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() {
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的消费
System.out.println(new String(msg.getBody()));
}else if (msg.getTags() != null
&&msg.getTags().equals("TagD")) {
//执行TagD的消费
System.out.println(new String(msg.getBody()));
}
}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("ConsumerStarted.");
}
}
启动消费者
Hello MetaQA
ConsumeMessageThread-ConsumerGroupName-8 Receive New Messages: 1
Hello MetaQB
ConsumeMessageThread-ConsumerGroupName-9 Receive New Messages: 1
Hello MetaQA
ConsumeMessageThread-ConsumerGroupName-10 Receive New Messages: 1
Hello MetaQB
ConsumeMessageThread-ConsumerGroupName-11 Receive New Messages: 1
Hello MetaQA
ConsumeMessageThread-ConsumerGroupName-12 Receive New Messages: 1
Hello MetaQB
ConsumeMessageThread-ConsumerGroupName-13 Receive New Messages: 1
Hello MetaQA
ConsumeMessageThread-ConsumerGroupName-14 Receive New Messages: 1
Hello MetaQB
ConsumeMessageThread-ConsumerGroupName-15 Receive New Messages: 1
Hello MetaQA
ConsumeMessageThread-ConsumerGroupName-16 Receive New Messages: 1
Hello MetaQB
ConsumeMessageThread-ConsumerGroupName-17 Receive New Messages: 1
Hello MetaQA
ConsumeMessageThread-ConsumerGroupName-18 Receive New Messages: 1
Hello MetaQB
ConsumeMessageThread-ConsumerGroupName-19 Receive New Messages: 1
Hello MetaQA
ConsumeMessageThread-ConsumerGroupName-20 Receive New Messages: 1
Hello MetaQB
运行生产者，
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002715, messageQueue=MessageQueue [topic=TopicTest2, brokerName=E97M3HS2ANMGCSZ, queueId=2], queueOffset=4]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F00000000000027AB, messageQueue=MessageQueue [topic=TopicTest3, brokerName=E97M3HS2ANMGCSZ, queueId=2], queueOffset=4]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002840, messageQueue=MessageQueue [topic=TopicTest1, brokerName=E97M3HS2ANMGCSZ, queueId=3], queueOffset=4]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F00000000000028D5, messageQueue=MessageQueue [topic=TopicTest2, brokerName=E97M3HS2ANMGCSZ, queueId=3], queueOffset=4]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F000000000000296B, messageQueue=MessageQueue [topic=TopicTest3, brokerName=E97M3HS2ANMGCSZ, queueId=3], queueOffset=4]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002A00, messageQueue=MessageQueue [topic=TopicTest1, brokerName=E97M3HS2ANMGCSZ, queueId=0], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002A95, messageQueue=MessageQueue [topic=TopicTest2, brokerName=E97M3HS2ANMGCSZ, queueId=0], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002B2B, messageQueue=MessageQueue [topic=TopicTest3, brokerName=E97M3HS2ANMGCSZ, queueId=0], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002BC0, messageQueue=MessageQueue [topic=TopicTest1, brokerName=E97M3HS2ANMGCSZ, queueId=1], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002C55, messageQueue=MessageQueue [topic=TopicTest2, brokerName=E97M3HS2ANMGCSZ, queueId=1], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002CEB, messageQueue=MessageQueue [topic=TopicTest3, brokerName=E97M3HS2ANMGCSZ, queueId=1], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002D80, messageQueue=MessageQueue [topic=TopicTest1, brokerName=E97M3HS2ANMGCSZ, queueId=2], queueOffset=5]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002E15, messageQueue=MessageQueue [topic=TopicTest2, brokerName=E97M3HS2ANMGCSZ, queueId=2], queueOffset=5]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002EAB, messageQueue=MessageQueue [topic=TopicTest3, brokerName=E97M3HS2ANMGCSZ, queueId=2], queueOffset=5]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002F40, messageQueue=MessageQueue [topic=TopicTest1, brokerName=E97M3HS2ANMGCSZ, queueId=3], queueOffset=5]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002FD5, messageQueue=MessageQueue [topic=TopicTest2, brokerName=E97M3HS2ANMGCSZ, queueId=3], queueOffset=5]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F000000000000306B, messageQueue=MessageQueue [topic=TopicTest3, brokerName=E97M3HS2ANMGCSZ, queueId=3], queueOffset=5]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000003100, messageQueue=MessageQueue [topic=TopicTest1, brokerName=E97M3HS2ANMGCSZ, queueId=0], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000003195, messageQueue=MessageQueue [topic=TopicTest2, brokerName=E97M3HS2ANMGCSZ, queueId=0], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F000000000000322B, messageQueue=MessageQueue [topic=TopicTest3, brokerName=E97M3HS2ANMGCSZ, queueId=0], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F00000000000032C0, messageQueue=MessageQueue [topic=TopicTest1, brokerName=E97M3HS2ANMGCSZ, queueId=1], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000003355, messageQueue=MessageQueue [topic=TopicTest2, brokerName=E97M3HS2ANMGCSZ, queueId=1], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F00000000000033EB, messageQueue=MessageQueue [topic=TopicTest3, brokerName=E97M3HS2ANMGCSZ, queueId=1], queueOffset=8]
http://www.soadmin.com/zonghe/open-source/1015259.htm
大家可以加我个人微信号：scccdgf
相关推荐
rocketmq-spring-boot-starter 阿里云RocketMQSpring图书版支持功能： 发送普通消息的三种模式：同步，异步和单向 订阅消息群集，广播 发送和接收顺序消息 交易讯息 延迟讯息 接收和接收定时消息定时消息和延迟消息...
< dependency> < group>com.qianmi</ group> < artifactId>spring-boot-starter-rocketmq</ artifactId> < version>1.1.0-RELEASE</ version></ dependency>产生讯息# # application.propertiesspring.rocketmq....
弹簧启动启动器rocketmq 帮助...< dependency> < groupId>org.apache.rocketmq</ groupId> < artifactId>spring-boot-starter-rocketmq</ artifactId> < version>1.0.0-SNAPSHOT</ version></ dependency>产生讯息 # #
带有Apollo Server的GraphQL服务器示例，Koa中间件，使用Knex JS的数据库读取和/或REST API使用者，令牌验证，与Redis和Nats的消息传递集成以及许多其他很酷的功能。 如果要创建新的Web应用程序，建议您使用 。 ...
讯息发布–子网站发布讯息Step1.pdf
人工智能时代“媒介即讯息”再审视.pdf
台湾电信业龙头中华电信因应网际网络与手机等...企业可透过中华电信的hiBox系统，整合语音及数据等多媒体信息为单一讯息信箱，并以网络、无线装置和市内电话传送与接收电子邮件和讯息，使企业运筹帷幄，不受时空限制。
Iris Wireless 公司选择 IBM BladeCenter 刀片服务器平台来支持和加速为其全球客户提供的无线讯息和高级移动数据服务。Iris Wireless 公司是一家提供先进的移动讯息和数据应用的全球性公司，在这次的项目中，IBM 的...
C++Builder处理Windows讯息,进行低层开发,需要用到讯息处理.
电信设备-半导体设备通信标准的讯息交换装置与讯息交换方法.zip
电信设备-具有自动发出讯息功能的移动通讯装置以及自动发出讯息的方法.zip
行业资料-交通装置-一种具有讯息接收和胎压侦测功能的车载装置.exe
第2讲信息、符号与讯息，第2讲信息、符号与讯息课件，第2讲信息、符号与讯息PPT
常见的HTTP错误讯息中文解读
类似MSN Messages的POP讯息框
传递网络安全讯息.pdf
认知发展讯息处理观点.ppt
讯息战略的目标方法.pptx
学习电子商务的讯息战网路行销.ppt
Panasonic-自我诊断错误讯息.doc