RocketMQ在windows上安装和eclipse开发使用
1.概述
RocketMQ是alibaba公司开源的一个纯java的开源消息中间件。
2.开发测试环境搭建
1. 安装&启动
进入到RocketMQ下载包解压的路径下
D:\machine\RocketMQ-3.0.8\RocketMQ-3.0.8>
接下来安装
执行下边的命令或者执行install.bat(在这个bat文件中的命令如下)对maven熟悉的一眼就知道是执行clean package install assembly等操作。
mvn -Dmaven.test.skip=true clean packageinstall assembly:assembly –U
操作信息如下:
[INFO] Scanning for projects... [INFO] ------------------------------------------------------------------------ [INFO] Reactor Build Order: [INFO] [INFO] rocketmq-all 3.0.8 [INFO] rocketmq-remoting 3.0.8 [INFO] rocketmq-common 3.0.8 [INFO] rocketmq-client 3.0.8 [INFO] rocketmq-store 3.0.8 [INFO] rocketmq-broker 3.0.8 [INFO] rocketmq-tools 3.0.8 [INFO] rocketmq-research 3.0.8 [INFO] rocketmq-namesrv 3.0.8 [INFO] rocketmq-example 3.0.8 [INFO] rocketmq-qatest 3.0.8 [INFO] [INFO] ------------------------------------------------------------------------ [INFO] Building rocketmq-all 3.0.8 3.0.8 [INFO]------------------------------------------------------------------------ [INFO]///省了…..编译打包过程 [INFO]------------------------------------------------------------------------ [INFO] Reactor Summary: [INFO] [INFO] rocketmq-all 3.0.8................................ SUCCESS [5.861s] [INFO] rocketmq-remoting 3.0.8 ...........................SUCCESS [1.483s] [INFO] rocketmq-common 3.0.8............................. SUCCESS [1.627s] [INFO] rocketmq-client 3.0.8............................. SUCCESS [1.486s] [INFO] rocketmq-store 3.0.8.............................. SUCCESS [1.398s] [INFO] rocketmq-broker 3.0.8............................. SUCCESS [1.512s] [INFO] rocketmq-tools 3.0.8.............................. SUCCESS [1.125s] [INFO] rocketmq-research 3.0.8........................... SUCCESS [0.917s] [INFO] rocketmq-namesrv 3.0.8............................ SUCCESS [0.609s] [INFO] rocketmq-example 3.0.8............................ SUCCESS [0.604s] [INFO] rocketmq-qatest 3.0.8............................. SUCCESS [0.057s] [INFO] ------------------------------------------------------------------------ [INFO] BUILD SUCCESS [INFO]------------------------------------------------------------------------ [INFO] Total time: 18.145s [INFO] Finished at: Fri Mar 28 10:36:08 CST2014 [INFO] Final Memory: 60M/395M [INFO]------------------------------------------------------------------------ D:\machine\RocketMQ-3.0.8\RocketMQ-3.0.8>接下来把编译好的项目copy出来
D:\machine\RocketMQ-3.0.8\RocketMQ-3.0.8\target> xcopy /E alibaba-rocketmq-3.0.8 D:\machine\RocketMQ-3.0.8\ D:\machine\RocketMQ-3.0.8>dir 驱动器 D 中的卷是 软件 卷的序列号是000F-1D4F D:\machine\RocketMQ-3.0.8 的目录 2014/03/28 10:44 <DIR> . 2014/03/28 10:44 <DIR> .. 2014/03/28 10:44 <DIR> alibaba-rocketmq 2014/03/07 15:08 52pax_global_header 2014/03/28 10:37 <DIR> RocketMQ-3.0.8 1 个文件 52 字节 4 个目录 105,101,885,440 可用字节 D:\machine\RocketMQ-3.0.8>cd alibaba-rocketmq启动服务
D:\machine\RocketMQ-3.0.8\alibaba-rocketmq>start/b bin/mqnamesrv.exe >D:\logs\alibaba-rocketmq/mqnamesrv.log可以通过jps查看一下是不是有了RocketMQ的进程,如下方的6484
C:\Users\houchangren>jps -v 6484 -Djava.ext.dirs=D:\machine\RocketMQ-3.0.8\alibaba-rocketmq\bin/../lib-Drocketmq.home.dir=D:\machine\RocketMQ-3.0.8\alibaba-rocketmq\bin/..-XX:MaxNewSize=512M -XX:MaxPermSize=128M -XX:NewSit abort 6876 JConsole -Denv.class.path=D:\ProgramFiles\Java\jdk1.6.0_26\lib -Dapplication.home=D:\Program Files\Java\jdk1.6.0_26-Djconsole.showOutputViewer 6936org.eclipse.equinox.launcher_1.2.0.v20110502.jar -Dosgi.requiredJavaVersion=1.5-Xms40m -Xmx512m -XX:MaxPermSize=256m 4140 Jps -Denv.class.path=D:\ProgramFiles\Java\jdk1.6.0_26\lib -Dapplication.home=D:\Program Files\Java\jdk1.6.0_26-Xms8m
看日志中信息D:\logs\alibaba-rocketmq/mqnamesrv.log
The Name Server boot success.
D:\machine\RocketMQ-3.0.8\alibaba-rocketmq>start/b bin/mqbroker.exe -n "10.57.41.19:9876">D:\logs\alibaba-rocketmq/mqbroker.log
2.项目实例
1. 创建maven项目
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.0http://maven.apache.org/maven-v4_0_0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.ruishenh</groupId> <artifactId>gomeTest</artifactId> <packaging>war</packaging> <version>0.0.1-SNAPSHOT</version> <name>gomeTest Maven Webapp</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.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> <build> <finalName>gomeTest</finalName> </build> </project>
2. 编写消息产生者Producer
文件路径:/gomeTest/src/main/java/com/ruishenh/rocketmq/example/Producer.java
package com.ruishenh.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 classProducer { public static void main(String[] args) throws MQClientException, InterruptedException{ /** * 一个应用创建一个Producer,由应用来维护此对象,可以设置为全局对象或者单例<br> * 注意:ProducerGroupName需要由应用来保证唯一<br> * ProducerGroup这个概念发送普通的消息时,作用不大,但是发送分布式事务消息时,比较关键, * 因为服务器会回查这个Group下的任意一个Producer */ final DefaultMQProducerproducer = newDefaultMQProducer("ProducerGroupName"); producer.setNamesrvAddr("10.57.41.19: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 { { Messagemsg = newMessage("TopicTest1",// topic "TagA",// tag "OrderID001",// key ("Hello MetaQA").getBytes());// body SendResultsendResult = producer.send(msg); System.out.println(sendResult); } { Messagemsg = newMessage("TopicTest2",// topic "TagB",// tag "OrderID0034",// key ("Hello MetaQB").getBytes());// body SendResultsendResult = producer.send(msg); System.out.println(sendResult); } { Messagemsg = newMessage("TopicTest3",// topic "TagC",// tag "OrderID061",// key ("Hello MetaQC").getBytes());// body SendResultsendResult = 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); } }
3. 编写消息消费者Consumer
文件路径:Test/src/main/java/com/ruishenh/rocketmq/example/PushConsumer.java
package com.ruishenh.rocketmq.example; import java.util.List; importcom.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer; importcom.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; importcom.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; importcom.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently; importcom.alibaba.rocketmq.client.exception.MQClientException; importcom.alibaba.rocketmq.common.message.MessageExt; public class PushConsumer { /** * 当前例子是PushConsumer用法,使用方式给用户感觉是消息从RocketMQ服务器推到了应用客户端。<br> * 但是实际PushConsumer内部是使用长轮询Pull方式从MetaQ服务器拉消息,然后再回调用户Listener方法<br> */ publicstatic void main(String[] args) throws InterruptedException, MQClientException{ /** * 一个应用创建一个Consumer,由应用来维护此对象,可以设置为全局对象或者单例<br> * 注意:ConsumerGroupName需要由应用来保证唯一 */ DefaultMQPushConsumerconsumer = new DefaultMQPushConsumer( "ConsumerGroupName"); consumer.setNamesrvAddr("10.57.41.19:9876"); consumer.setInstanceName("Consumber"); /** * 订阅指定topic下tags分别等于TagA或TagC或TagD */ consumer.subscribe("TopicTest1","TagA || TagC || TagD"); /** * 订阅指定topic下所有消息<br> * 注意:一个consumer对象可以订阅多个topic */ consumer.subscribe("TopicTest2","*"); consumer.registerMessageListener(newMessageListenerConcurrently() { publicConsumeConcurrentlyStatus consumeMessage( List<MessageExt>msgs, ConsumeConcurrentlyContext context) { System.out.println(Thread.currentThread().getName() +" Receive New Messages: " + msgs.size()); MessageExtmsg = msgs.get(0); if(msg.getTopic().equals("TopicTest1")) { //执行TopicTest1的消费逻辑 if(msg.getTags() != null && msg.getTags().equals("TagA")) { //执行TagA的消费 System.out.println(newString(msg.getBody())); }else if (msg.getTags() != null &&msg.getTags().equals("TagC")) { //执行TagC的消费 System.out.println(newString(msg.getBody())); }else if (msg.getTags() != null &&msg.getTags().equals("TagD")) { //执行TagD的消费 System.out.println(newString(msg.getBody())); } }else if (msg.getTopic().equals("TopicTest2")) { System.out.println(newString(msg.getBody())); } returnConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); /** * Consumer对象在使用之前必须要调用start初始化,初始化一次即可<br> */ consumer.start(); System.out.println("ConsumerStarted."); } }
参考:
http://my.oschina.net/cloudcoder/blog/200741https://github.com/alibaba/RocketMQ/wiki/Quick-Start
相关推荐
LCFlxfldy 2020-08-17
IT农场 2020-11-13
ljcsdn 2020-07-27
LCFlxfldy 2020-07-05
lypgcs 2020-06-27
陈晨软件五千言 2020-06-17
qingyuerji 2020-06-14
MojitoBlogs 2020-06-14
lypgcs 2020-06-14
陈晨软件五千言 2020-06-14
meilongwhpu 2020-06-13
陈晨软件五千言 2020-06-11
qingyuerji 2020-06-09
MojitoBlogs 2020-06-09
meilongwhpu 2020-06-08
meilongwhpu 2020-06-08
lypgcs 2020-06-07
MojitoBlogs 2020-06-04
meilongwhpu 2020-05-30