`

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

    博客分类：
  • J2EE
 
阅读更多

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
分享到：
| 生成校验码
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    Windows上Eclipse搭建Android开发环境教程

    本文旨在指导读者在 Windows 平台上搭建 Android 开发环境，涵盖了 JDK 的安装、配置 Windows 上 JDK 的变量环境、下载安装 Eclipse、下载安装 Android SDK 和为 Eclipse 安装 ADT 插件等五个步骤。 知识点一：JDK ...

    如何在windows环境下安装Eclipse的C++开发工具

    以下是如何在Windows上安装Eclipse C++开发工具的详细步骤： 首先，你需要下载Eclipse IDE for C/C++ Developers。访问官方网站（http://www.eclipse.org/downloads/）并选择适合你系统的版本，例如Windows 32位。...

    windows+eclipse+cdt开发环境配置

    至此，你已经成功在Windows平台上配置好了Eclipse CDT开发环境，可以开始编写和编译C/C++代码了。通过以上步骤，不仅能够确保项目的顺利编译和运行，还能够利用Eclipse强大的编辑和调试功能，提高开发效率。 ### ...

    eclipse安装Hadoop插件

    1. **Eclipse IDE**：确保已安装最新版本的Eclipse IDE，推荐使用Eclipse Oxygen或更高版本。 2. **Hadoop环境**：已安装好Hadoop环境，并且能够正常启动Hadoop服务。 3. **JDK环境**：确保计算机上已正确安装JDK，...

    分布详解——搭建Windows下基于Eclipse的PHP开发环境

    在本文中，我们将深入探讨如何在Windows操作系统上搭建一个基于Eclipse的PHP开发环境。Eclipse是一款强大的开源集成开发环境（IDE），它支持多种编程语言，包括PHP。通过安装特定的PHP插件，Eclipse可以成为PHP...

    中标麒麟系统上安装Java、Eclipse指南.doc

    中标麒麟系统上安装Java、Eclipse指南是指在国产操作系统中标麒麟系统上安装Java开发环境和Eclipse集成开发环境的指南。该指南提供了详细的安装步骤和配置过程，旨在帮助用户快速安装和配置Java和Eclipse环境，方便...

    windows下Eclipse开发环境的搭建

    5. **安装ADT插件**：在Eclipse中，通过`Help` &gt; `Software Updates` &gt; `Find and Install` &gt; `Search for new features to install`，然后输入远程站点名称和URL（https://dl-ssl.google.com/android/eclipse/），...

    eclipse中安装Mybatipse与使用指南

    在Java开发领域，Eclipse是一款广泛使用的集成开发环境（IDE），尤其在处理Spring、Spring MVC和MyBatis（简称SSM）这样的企业级框架时。为了提升开发效率，开发者经常需要借助各种插件来增强Eclipse的功能。...

    在windows上建立hadoop+eclipse开发环境

    在Windows上搭建Hadoop+Eclipse开发环境是一项重要的工作，它涉及到多个步骤，包括JDK、Cygwin、SSHD服务、Hadoop的安装与配置，以及Eclipse插件的安装。以下是对这些步骤的详细说明： 1. **安装JDK**：首先需要...

    Eclipse和PyDev搭建完美Python开发环境（Windows篇）

    在Windows平台上，使用Eclipse和PyDev构建一个高效的Python开发环境是许多程序员的选择。Eclipse是一款功能强大的开源集成开发环境（IDE），而PyDev是专为Eclipse设计的一个Python插件，提供了丰富的Python开发功能...

    在eclipse中安装和使用cvs

    本课件介绍了在eclipse开发环境下安装和使用cvs插件的方法。cvs用于联合开发管理，在共同开发一个工程时使用cvs可以保证小组成员之间的协作和同步。

    使用jdk、eclipse、MinGw在windows环境下搭建linux环境进行c/c++开发

    ### 使用JDK、Eclipse、MinGw在Windows环境下搭建Linux环境进行C/C++开发的知识点 #### 一、概述 在Windows环境下搭建一个能够支持C/C++开发的环境，通常需要安装一系列的软件和配置相应的环境变量。本文将详细介绍...

    在Windows操作系统上安装Eclipse

    Eclipse 在 Windows 操作系统上的安装步骤 Eclipse 是一个流行的集成开发环境（IDE），它支持多种编程语言，包括 Java、Python、C++ 等。要在 Windows 操作系统上安装 Eclipse，需要按照以下步骤进行操作。 知识点...

    NC的Eclipse开发插件

    3. **NC二次开发**：二次开发通常是指在已有软件基础上进行的定制化开发，以满足特定用户或业务的需求。对于NC系统而言，二次开发可能涉及界面修改、功能增强、数据接口对接等方面。 4. **插件集成**：将NC的...

    Windows 下基于 Eclipse 的可视化远程 Linux CC++ 开发环境搭建

    6. **启动Eclipse**：在Linux服务器上，使用`nohup eclipse &gt; /dev/null &`命令启动Eclipse，此时Eclipse的界面将在Windows上的Xming中显示。 这样的开发环境能够提供一个高效的开发流程，包括代码编辑、编译和调试...

    如何使用Eclipse来开发Android

    综上所述，学习如何在Eclipse中开发Android应用需要理解Eclipse的基本操作，安装和配置必要的Android开发工具，以及掌握Android应用的基本结构和生命周期。尽管现代开发者更倾向于使用Android Studio，但了解Eclipse...

Magicbox
Global site tag (gtag.js) - Google Analytics