MyException - 我的异常网
当前位置:我的异常网» Windows » RocketMQ在windows上安装跟eclipse开发使用

RocketMQ在windows上安装跟eclipse开发使用

www.MyException.Cn  网友分享于:2015-08-26  浏览:0次
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/200741
https://github.com/alibaba/RocketMQ/wiki/Quick-Start

文章评论

程序猿的崛起——Growth Hacker
程序猿的崛起——Growth Hacker
当下全球最炙手可热的八位少年创业者
当下全球最炙手可热的八位少年创业者
程序员和编码员之间的区别
程序员和编码员之间的区别
Java程序员必看电影
Java程序员必看电影
为什么程序员都是夜猫子
为什么程序员都是夜猫子
10个调试和排错的小建议
10个调试和排错的小建议
旅行,写作,编程
旅行,写作,编程
鲜为人知的编程真相
鲜为人知的编程真相
编程语言是女人
编程语言是女人
Web开发人员为什么越来越懒了?
Web开发人员为什么越来越懒了?
程序员应该关注的一些事儿
程序员应该关注的一些事儿
团队中“技术大拿”并非越多越好
团队中“技术大拿”并非越多越好
每天工作4小时的程序员
每天工作4小时的程序员
为啥Android手机总会越用越慢?
为啥Android手机总会越用越慢?
Web开发者需具备的8个好习惯
Web开发者需具备的8个好习惯
10个帮程序员减压放松的网站
10个帮程序员减压放松的网站
5款最佳正则表达式编辑调试器
5款最佳正则表达式编辑调试器
2013年美国开发者薪资调查报告
2013年美国开发者薪资调查报告
程序员的一天:一寸光阴一寸金
程序员的一天:一寸光阴一寸金
我的丈夫是个程序员
我的丈夫是个程序员
Google伦敦新总部 犹如星级庄园
Google伦敦新总部 犹如星级庄园
老程序员的下场
老程序员的下场
如何成为一名黑客
如何成为一名黑客
程序员都该阅读的书
程序员都该阅读的书
我跳槽是因为他们的显示器更大
我跳槽是因为他们的显示器更大
如何区分一个程序员是“老手“还是“新手“?
如何区分一个程序员是“老手“还是“新手“?
老美怎么看待阿里赴美上市
老美怎么看待阿里赴美上市
十大编程算法助程序员走上高手之路
十大编程算法助程序员走上高手之路
60个开发者不容错过的免费资源库
60个开发者不容错过的免费资源库
程序员最害怕的5件事 你中招了吗?
程序员最害怕的5件事 你中招了吗?
程序员周末都喜欢做什么?
程序员周末都喜欢做什么?
亲爱的项目经理,我恨你
亲爱的项目经理,我恨你
程序员必看的十大电影
程序员必看的十大电影
那些性感的让人尖叫的程序员
那些性感的让人尖叫的程序员
“肮脏的”IT工作排行榜
“肮脏的”IT工作排行榜
代码女神横空出世
代码女神横空出世
什么才是优秀的用户界面设计
什么才是优秀的用户界面设计
程序员眼里IE浏览器是什么样的
程序员眼里IE浏览器是什么样的
漫画:程序员的工作
漫画:程序员的工作
那些争议最大的编程观点
那些争议最大的编程观点
要嫁就嫁程序猿—钱多话少死的早
要嫁就嫁程序猿—钱多话少死的早
总结2014中国互联网十大段子
总结2014中国互联网十大段子
Java 与 .NET 的平台发展之争
Java 与 .NET 的平台发展之争
看13位CEO、创始人和高管如何提高工作效率
看13位CEO、创始人和高管如何提高工作效率
程序员的鄙视链
程序员的鄙视链
聊聊HTTPS和SSL/TLS协议
聊聊HTTPS和SSL/TLS协议
写给自己也写给你 自己到底该何去何从
写给自己也写给你 自己到底该何去何从
 程序员的样子
程序员的样子
一个程序员的时间管理
一个程序员的时间管理
软件开发程序错误异常ExceptionCopyright © 2009-2015 MyException 版权所有