淘宝消息中间件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应用程序,建议您使用 。 ...
本文件"半导体设备通信标准的讯息交换装置与讯息交换方法"详细阐述了这一主题,旨在提高通信系统的性能和兼容性。以下是对该主题的深入解析: 1. **通信标准**:通信标准是确保不同设备之间能进行无缝交互的一套...
在这个“08-‘我讯息’分析情绪案例”中,我们将探讨如何使用“我讯息”来理解和处理情绪困扰。 首先,案例中的主角经历了作为小组代表上台发言失败的挫折,导致他在下次团日活动汇报前感到压力和焦虑,甚至因此...
此外,用户权限管理也是重要的安全措施,只有经过授权的应用才能访问和使用发送消息的接口。 在实际应用中,这种自动发送消息的功能还可以结合云服务进行扩展,例如,通过物联网(IoT)平台,远程设备可以实时上传...
台达VFD-S系列简易功能操作型变频器是一种应用于交流电机驱动的设备,它采用了高品质元件和先进的微电脑控制技术。以下是从标题、描述和部分内容中提取的关键知识点: 1. 使用前注意事项: - 在安装和操作变频器前...
手册涵盖了安装前提条件、安装步骤、网络连接和零客户机的初始设定等内容。 一、PCoIP 主机卡简介 Tera 2220_PCoIP 主机卡是一款高性能的 PCoIP 主机卡,支持双屏或四屏显示,可以与零客户机配合使用,提供高质量的...
【标题】"讯息发布–子网站发布讯息Step1.pdf"主要介绍的是一个关于在子网站上发布信息的步骤,特别关注了消息发布的初步流程。这个文档可能是一份操作指南,旨在帮助用户或者管理员有效地在子网站上分享和传播信息...
在电子商务的讯息战网路行销中,STP分析是非常重要的一步,STP是Segmentation、Targeting和Positioning的缩写,分别对应市场分割、目标市场选择和市场定位。在网络营销策略的发展中,STP分析是极其重要的一步,因为...
#### 一、4xx: Client Error 使用者端(浏览器)错误讯息 这部分错误通常是因为客户端(如浏览器)发送了错误的请求或请求不符合服务器的期望导致的。 - **400 BadRequest**:错误的要求。服务器无法理解客户端...
标题中的“类似MSN Messages的POP讯息框”指的是在计算机通信和网络应用中,一种模拟微软MSN Messenger(现已被Skype取代)消息提示效果的POP(Point of Presence)消息框。这种消息框通常用于实现即时通讯软件或者...
需要设置软件源,以便能够下载和安装软件。可以选择使用官方软件源或第三方软件源。 2. 安装软件 可以使用 apt-get 命令来安装软件,例如:sudo apt-get install firefox 基本配置 1. 设置时间和日期 需要设置时间...
讯息系统是Windows操作系统的基础组成部分,它允许应用程序通过消息传递机制与系统和其他应用程序进行交互。本讲座深入探讨了如何使用Visual Basic(VB)编程语言来操作Windows API,特别强调了消息队列、窗口过程...
在本项目中,“C/C++ 学生讯息管理同学录”是一个初学者编写的程序,用于管理和存储学生的信息。这个程序可能包含了基本的学生数据结构、文件操作以及用户交互等功能,帮助用户添加、查看和更新学生信息。让我们深入...
CATIA 许可证服务安装是指安装和配置 CATIA 许可证服务,以便让用户能够正确地使用 CATIA 软件。在这个过程中,我们将详细介绍安装和配置 CATIA 许可证服务的步骤。 一、安装 License Use Management(LUM) Runtime ...
6. **错误处理和调试**:学习如何使用On Error语句进行错误处理,以及使用调试工具来查找和修复问题。 7. **状态管理**:通过Session和Application对象来管理用户的会话状态和全局应用状态。 8. **URL重写**:可能...