`
wbj0110
  • 浏览: 1604565 次
  • 性别: Icon_minigender_1
  • 来自: 上海
文章分类
社区版块
存档分类
最新评论

淘宝讯息中间件RocketMQ的安装和简单使用

RocketMQ 
阅读更多

淘宝消息中间件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

 

 

微信公众号：
分享到：
| Cloudera Manager 和 CDH 4 终极安
评论
发表评论

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

相关推荐

    rocketmq-spring-boot-starter：rocketmq-spring-boot-starter

    rocketmq-spring-boot-starter 阿里云RocketMQSpring图书版支持功能： 发送普通消息的三种模式：同步，异步和单向 订阅消息群集，广播 发送和接收顺序消息 交易讯息 延迟讯息 接收和接收定时消息定时消息和延迟消息...

    spring-boot-starter-rocketmq：使用Apache RocketMQ进行消息传递的启动器

    &lt; dependency&gt; &lt; group&gt;com.qianmi&lt;/ group&gt; &lt; artifactId&gt;spring-boot-starter-rocketmq&lt;/ artifactId&gt; &lt; version&gt;1.1.0-RELEASE&lt;/ version&gt;&lt;/ dependency&gt;产生讯息# # application.propertiesspring.rocketmq....

    人工智能时代“媒介即讯息”再审视.pdf

    麦克卢汉认为媒介包括一切技术和人对技术的使用，这些技术延伸了人的感知能力，比如印刷品是眼睛的延伸。这种泛媒介观与媒介环境学派的观点一脉相承，即媒介在人类社会和文化的发展中起到了决定性的作用。 文章...

    spring-boot-starter-rocketmq

    弹簧启动启动器rocketmq 帮助...&lt; dependency&gt; &lt; groupId&gt;org.apache.rocketmq&lt;/ groupId&gt; &lt; artifactId&gt;spring-boot-starter-rocketmq&lt;/ artifactId&gt; &lt; version&gt;1.0.0-SNAPSHOT&lt;/ version&gt;&lt;/ dependency&gt;产生讯息 # #

    generator-graphql-rocket:带有Apollo Server的GraphQL服务器示例，Koa中间件，使用Knex JS和_或REST API使用者的数据库读取，令牌验证，与Redis和Nats的消息传递集成以及许多其他很酷的功能

    带有Apollo Server的GraphQL服务器示例，Koa中间件，使用Knex JS的数据库读取和/或REST API使用者，令牌验证，与Redis和Nats的消息传递集成以及许多其他很酷的功能。 如果要创建新的Web应用程序，建议您使用 。 ...

    电信设备-半导体设备通信标准的讯息交换装置与讯息交换方法.zip

    本文件"半导体设备通信标准的讯息交换装置与讯息交换方法"详细阐述了这一主题，旨在提高通信系统的性能和兼容性。以下是对该主题的深入解析： 1. **通信标准**：通信标准是确保不同设备之间能进行无缝交互的一套...

    08-“我讯息”分析情绪案例PPT·.pptx

    在这个“08-‘我讯息’分析情绪案例”中，我们将探讨如何使用“我讯息”来理解和处理情绪困扰。 首先，案例中的主角经历了作为小组代表上台发言失败的挫折，导致他在下次团日活动汇报前感到压力和焦虑，甚至因此...

    电信设备-具有自动发出讯息功能的移动通讯装置以及自动发出讯息的方法.zip

    此外，用户权限管理也是重要的安全措施，只有经过授权的应用才能访问和使用发送消息的接口。 在实际应用中，这种自动发送消息的功能还可以结合云服务进行扩展，例如，通过物联网(IoT)平台，远程设备可以实时上传...

    台达 VFD-S 简易功能操作型 变频器 使用手册.pdf

    台达VFD-S系列简易功能操作型变频器是一种应用于交流电机驱动的设备，它采用了高品质元件和先进的微电脑控制技术。以下是从标题、描述和部分内容中提取的关键知识点： 1. 使用前注意事项： - 在安装和操作变频器前...

    Tera 2220_PCoIP_Host_Card_快速安装手册

    手册涵盖了安装前提条件、安装步骤、网络连接和零客户机的初始设定等内容。 一、PCoIP 主机卡简介 Tera 2220_PCoIP 主机卡是一款高性能的 PCoIP 主机卡，支持双屏或四屏显示，可以与零客户机配合使用，提供高质量的...

    讯息发布–子网站发布讯息Step1.pdf

    【标题】"讯息发布–子网站发布讯息Step1.pdf"主要介绍的是一个关于在子网站上发布信息的步骤，特别关注了消息发布的初步流程。这个文档可能是一份操作指南，旨在帮助用户或者管理员有效地在子网站上分享和传播信息...

    学习电子商务的讯息战网路行销.ppt

    在电子商务的讯息战网路行销中，STP分析是非常重要的一步，STP是Segmentation、Targeting和Positioning的缩写，分别对应市场分割、目标市场选择和市场定位。在网络营销策略的发展中，STP分析是极其重要的一步，因为...

    常见的HTTP错误讯息中文解读.docx

    #### 一、4xx: Client Error 使用者端（浏览器）错误讯息 这部分错误通常是因为客户端（如浏览器）发送了错误的请求或请求不符合服务器的期望导致的。 - **400 BadRequest**：错误的要求。服务器无法理解客户端...

    类似MSN Messages的POP讯息框

    标题中的“类似MSN Messages的POP讯息框”指的是在计算机通信和网络应用中，一种模拟微软MSN Messenger（现已被Skype取代）消息提示效果的POP（Point of Presence）消息框。这种消息框通常用于实现即时通讯软件或者...

    Ubuntu_10.10_图形安装教程、网络设置、软件源、基本配置.docx

    需要设置软件源，以便能够下载和安装软件。可以选择使用官方软件源或第三方软件源。 2. 安装软件 可以使用 apt-get 命令来安装软件，例如：sudo apt-get install firefox 基本配置 1. 设置时间和日期 需要设置时间...

    VB与WindowsAPI讲座叁Windows的讯息系统.pdf

    讯息系统是Windows操作系统的基础组成部分，它允许应用程序通过消息传递机制与系统和其他应用程序进行交互。本讲座深入探讨了如何使用Visual Basic（VB）编程语言来操作Windows API，特别强调了消息队列、窗口过程...

    C/C++ 学生讯息管理同学录

    在本项目中，“C/C++ 学生讯息管理同学录”是一个初学者编写的程序，用于管理和存储学生的信息。这个程序可能包含了基本的学生数据结构、文件操作以及用户交互等功能，帮助用户添加、查看和更新学生信息。让我们深入...

    CATIA许可证服务安装

    CATIA 许可证服务安装是指安装和配置 CATIA 许可证服务，以便让用户能够正确地使用 CATIA 软件。在这个过程中，我们将详细介绍安装和配置 CATIA 许可证服务的步骤。 一、安装 License Use Management(LUM) Runtime ...

    讯息内容管理系统 v2.1-ASP源码.zip

    6. **错误处理和调试**：学习如何使用On Error语句进行错误处理，以及使用调试工具来查找和修复问题。 7. **状态管理**：通过Session和Application对象来管理用户的会话状态和全局应用状态。 8. **URL重写**：可能...

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