- 浏览: 211325 次
- 性别:
- 来自: 杭州
文章分类
最新评论
-
hus20120405:
很不错,我运行起来了
log4jdbc日志框架介绍 -
Roshomon:
牛叉 找了很多解决方案 你的OK了
plsql ora-12154 TNS 无法处理服务名 -
ackley:
赞!谢谢!
Cannot access NLS data files or invalid ...
前段时间在实施分布式数据库的过程中,使用过rocketMQ, 因为搭建过程实在太顺利,当时忘了留文档,今天有同事来向我请教rocketMQ使用时,居然已经找不出当时的demo了 。
于是在收藏夹里重新翻出原来的学习内容,干脆写一篇学习笔记,加深映象吧。
1.下载最新版 rocketMq.
[urlhttps://github.com/alibaba/RocketMQ/releases/download/v3.2.6/alibaba-rocketmq-3.2.6.tar.gz[/url]
2.解压,并进入bin目录,运行启动命令:
启动nameserver: Start /b mqnamesrv.exe > mqnamesrv.log
启动broker: Start /b mqbroker.exe –n “192.168.180.15:9876” > mqbroker.log
3.搭建测试工程,引进jar包
<dependency>
<groupId>com.alibaba.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>3.2.6</version>
</dependency>
<dependency>
<groupId>com.alibaba.rocketmq</groupId>
<artifactId>rocketmq-tools</artifactId>
<version>3.2.6</version>
</dependency>
<dependency>
<groupId>com.alibaba.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
<version>3.2.6</version>
<type>pom</type>
</dependency>
4. 编写测试代码
发送端
消费端
package cn.gov.zjport.rocketmq.example;
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 PushConsumer {
/**
* 当前例子是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("192.168.180.15: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.");
}
}
于是在收藏夹里重新翻出原来的学习内容,干脆写一篇学习笔记,加深映象吧。
1.下载最新版 rocketMq.
[urlhttps://github.com/alibaba/RocketMQ/releases/download/v3.2.6/alibaba-rocketmq-3.2.6.tar.gz[/url]
2.解压,并进入bin目录,运行启动命令:
启动nameserver: Start /b mqnamesrv.exe > mqnamesrv.log
启动broker: Start /b mqbroker.exe –n “192.168.180.15:9876” > mqbroker.log
3.搭建测试工程,引进jar包
<dependency>
<groupId>com.alibaba.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>3.2.6</version>
</dependency>
<dependency>
<groupId>com.alibaba.rocketmq</groupId>
<artifactId>rocketmq-tools</artifactId>
<version>3.2.6</version>
</dependency>
<dependency>
<groupId>com.alibaba.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
<version>3.2.6</version>
<type>pom</type>
</dependency>
4. 编写测试代码
发送端
package cn.gov.zjport.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 class Producer { /** * 主函数 * @param args 参数 * @throws MQClientException * @throws InterruptedException */ public static void main(String[] args) throws MQClientException, InterruptedException{ /** * 一个应用创建一个Producer,由应用来维护此对象,可以设置为全局对象或者单例<br> * 注意:ProducerGroupName需要由应用来保证唯一<br> * ProducerGroup这个概念发送普通的消息时,作用不大,但是发送分布式事务消息时,比较关键, * 因为服务器会回查这个Group下的任意一个Producer */ final DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName"); producer.setNamesrvAddr("192.168.180.15: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方法 */ Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() { public void run() { producer.shutdown(); } })); System.exit(0); } }
消费端
package cn.gov.zjport.rocketmq.example;
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 PushConsumer {
/**
* 当前例子是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("192.168.180.15: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.");
}
}
发表评论
文章已被作者锁定,不允许评论。
-
zookeeper docker 集群安装
2022-10-14 17:18 1311. 下载镜像 docker pull docker.io/z ... -
Kafka学习笔记【三】-JAVA调用示例
2018-05-29 08:35 5691. pom.xml <dependency> ... -
Kafka学习笔记【二】-安装
2018-05-24 08:20 354kafka的安装如下: 一、准备环境 1) 三台服务器 ... -
Kafka学习笔记【一】-原理
2018-05-18 09:23 249一、kafka的重要概念 1. kafka 分为 Produ ... -
rabbitMQ学习笔记【三】-haproxy部署 原创
2018-04-11 16:11 374一、下载 https://fossies.org/lin ... -
rabbitMQ学习笔记【二】-集群部署 原创
2018-04-03 19:40 397一、准备工作 三台机器hosts: 192.168.3 ... -
rabbitMQ学习笔记【一】-单机部署 原创
2018-03-19 16:04 405公司对外接入需要支持MQ,最终选定rabbitMQ.本文 ... -
HttpClient 学习笔记【原创】
2017-09-07 19:28 1868HttpClient 某些方法的作用梳理。本文的测试基于htt ... -
关于dubbo的思考【原创】
2016-09-21 08:23 1314最近在看dubbo的文档,有些内容结合自己的思考记录一下 1. ... -
dubbo http webservice 服务学习笔记【原创】
2016-09-11 19:12 6470学习了 dubbo rest服务之后,再使用dubbo htt ... -
dubbo rest 服务学习笔记(三)【原创】
2016-09-09 19:34 8741在配置过程中遇到两个 ... -
dubbo rest 服务学习笔记(二)【原创】
2016-09-09 19:21 1374采用dubbo客户端访问dubbo发布的rest服务,接上一篇 ... -
dubbo rest 服务学习笔记(一)【原创】
2016-09-09 19:07 6065dubbo是很好的服务治理工具,包含了注册,路由,监控 ... -
RocketMq数据处理
2016-06-27 08:58 607这篇讲实践中数据丢失、顺序幂等处理、性能压测,有时间再写 -
RocketMq学习笔记(2)【原创】
2016-06-27 08:55 1122前一篇文章http://zhenggm.i ... -
MS MQ 使用实践【原创】
2016-06-26 14:58 7631. 参照百度经验http://jingyan.baidu.c ... -
rest的调用 java 示例代码【原创】
2016-06-20 15:35 11051. 服务端示例代码(如何搭建rest服务,请看上一篇) ... -
jersey2 与 spring4 整合实践【原创】
2016-06-13 15:35 4963前几年用过jeysey,那都是1.x版本的,jersey2有很 ... -
java nio入门
2010-03-04 17:22 952首先了解下所谓的java nio是个什么东西! ... -
spring http invoker学习笔记
2010-01-04 16:58 3659概念 Spring HTTP invoker是spring框架 ...
相关推荐
RocketMQ学习笔记 1 RocketMQ学习笔记 1是关于Apache RocketMQ的学习笔记,涵盖了RocketMQ的基本概念、架构、消息模型、Producer和Consumer的基本概念、JMS规范等知识点。 1. 消息模型 在RocketMQ中,消息模型是...
RocketMQ学习笔记主要涵盖了消息队列(MQ)的相关知识,内容深入浅出,不仅包含理论知识,还融入了实战案例和源码分析,适用于希望掌握RocketMQ的开发者。 首先,RocketMQ是一个分布式、开源的消息中间件,它起源于...
RocketMQ学习笔记1 RocketMQ是Apache旗下的一个开源的消息队列系统,具有分布式、可靠、可扩展、高性能等特点。下面是对RocketMQ的学习笔记的总结。 分布式架构 RocketMQ原生支持分布式,解决了单点故障问题,...
rocketmq学习 rocketmq深入
尚硅谷提供的 RocketMQ 学习视频笔记旨在帮助初学者系统地掌握 RocketMQ 的核心概念和使用方法。 1. **MQ 简述** 消息队列(Message Queue,MQ)是一种异步通信机制,它在生产者和消费者之间充当了缓冲的角色,...
RocketMq 学习笔记 包含集群配置、顺序消费、事务、Filter组件配置以及使用,包含demo
"RocketMQ黑马笔记" RocketMQ是阿里巴巴开源的消息中间件,在阿里内部历经了双十一等很多高并发场景的考验,能够处理亿万级别的消息。2016年开源后捐赠给Apache,现在是Apache的一个顶级项目。目前RocketMQ在阿里云...
Docker Desktop下安装rocketmq
rocketmq自用笔记第一部分
Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习...
希沃白板学习笔记.pdf希沃白板学习笔记.pdf希沃白板学习笔记.pdf希沃白板学习笔记.pdf希沃白板学习笔记.pdf希沃白板学习笔记.pdf希沃白板学习笔记.pdf希沃白板学习笔记.pdf希沃白板学习笔记.pdf
Springcloud学习笔记.md,Springcloud学习笔记.md,Springcloud学习笔记.md,Springcloud学习笔记.md,Springcloud学习笔记.md,Springcloud学习笔记.md,Springcloud学习笔记.md,Springcloud学习笔记.md,Spring...
人工智能学习笔记,人工智能学习笔记,人工智能学习笔记人工智能学习笔记,人工智能学习笔记,人工智能学习笔记人工智能学习笔记,人工智能学习笔记,人工智能学习笔记人工智能学习笔记,人工智能学习笔记,人工智能...
CCNA学习笔记 CCNA学习笔记 CCNA学习笔记
云的学习笔记-云的学习笔记系统-云的学习笔记系统源码-云的学习笔记管理系统-云的学习笔记管理系统java代码-云的学习笔记系统设计与实现-基于ssm的云的学习笔记系统-基于Web的云的学习笔记系统设计与实现-云的学习...
云的学习笔记-云的学习笔记系统-云的学习笔记系统源码-云的学习笔记管理系统-云的学习笔记管理系统java代码-云的学习笔记系统设计与实现-基于ssm的云的学习笔记系统-基于Web的云的学习笔记系统设计与实现-云的学习...