上一期介绍了我们项目要用到activeMQ来作为jms总线,并且给大家介绍了activeMQ的集群和高可用部署方案,本期给大家再介绍下,如何根据自己的项目需求,更好地使用activeMQ的两种消息处理模式。
1 queue与topic的技术特点对比
对比项 |
Topic |
Queue |
概要 |
Publish Subscribe messaging 发布订阅消息 |
Point-to-Point 点对点 |
有无状态 |
topic数据默认不落地,是无状态的。 |
Queue数据默认会在mq服务器上以文件形式保存,比如Active MQ一般保存在$AMQ_HOME\data\kr-store\data下面。也可以配置成DB存储。 |
完整性保障 |
并不保证publisher发布的每条数据,Subscriber都能接受到。 |
Queue保证每条数据都能被receiver接收。 |
消息是否会丢失 |
一般来说publisher发布消息到某一个topic时,只有正在监听该topic地址的sub能够接收到消息;如果没有sub在监听,该topic就丢失了。 |
Sender发送消息到目标Queue,receiver可以异步接收这个Queue上的消息。Queue上的消息如果暂时没有receiver来取,也不会丢失。 |
消息发布接收策略 |
一对多的消息发布接收策略,监听同一个topic地址的多个sub都能收到publisher发送的消息。Sub接收完通知mq服务器 |
一对一的消息发布接收策略,一个sender发送的消息,只能有一个receiver接收。receiver接收完后,通知mq服务器已接收,mq服务器对queue里的消息采取删除或其他操作。 |
Topic和queue的最大区别在于topic是以广播的形式,通知所有在线监听的客户端有新的消息,没有监听的客户端将收不到消息;而queue则是以点对点的形式通知多个处于监听状态的客户端中的一个。
2 topic和queue方式的消息处理效率比较
通过增加监听客户端的并发数来验证,topic的消息推送,是否会因为监听客户端的并发上升而出现明显的下降,测试环境的服务器为ci环境的ActiveMQ,客户端为我的本机。
从实测的结果来看,topic方式发送的消息,发送和接收的效率,在一个订阅者和100个订阅者的前提下没有明显差异,但在500个订阅者(线程)并发的前提下,效率差异很明显(由于500线程并发的情况下,我本机的cpu占用率已高达70-90%,所以无法确认是我本机测试造成的性能瓶颈还是topic消息发送方式存在性能瓶颈,造成效率下降如此明显)。
Topic方式发送的消息与queue方式发送的消息,发送和接收的效率,在一个订阅者和100个订阅者的前提下没有明显差异,但在500个订阅者并发的前提下,topic方式的效率明显低于queue。
Queue方式发送的消息,在一个订阅者、100个订阅者和500个订阅者的前提下,发送和接收的效率没有明显变化。
Topic实测数据:
|
发送者发送的消息总数 |
所有订阅者接收到消息的总数 |
消息发送和接收平均耗时 |
单订阅者 |
100 |
100 |
101ms |
100订阅者 |
100 |
10000 |
103ms |
500订阅者 |
100 |
50000 |
14162ms |
Queue实测数据:
|
发送者发送的消息总数 |
所有订阅者接收到消息的总数 |
消息发送和接收平均耗时 |
单订阅者 |
100 |
100 |
96ms |
100订阅者 |
100 |
100 |
96ms |
500订阅者 |
100 |
100 |
100ms |
3 topic方式的消息处理示例
3.1 通过客户端代码调用来发送一个topic的消息:
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
publicclass SendTopic {
privatestaticfinalintSEND_NUMBER = 5;
publicstaticvoid sendMessage(Session session, MessageProducer producer)
throws Exception {
for (int i = 1; i <=SEND_NUMBER; i++) {
TextMessage message = session
.createTextMessage("ActiveMq发送的消息" + i);
//发送消息到目的地方
System.out.println("发送消息:" + "ActiveMq 发送的消息" + i);
producer.send(message);
}
}
publicstaticvoid main(String[] args) {
// ConnectionFactory:连接工厂,JMS用它创建连接
ConnectionFactory connectionFactory;
// Connection:JMS客户端到JMS Provider的连接
Connection connection = null;
// Session:一个发送或接收消息的线程
Session session;
// Destination:消息的目的地;消息发送给谁.
Destination destination;
// MessageProducer:消息发送者
MessageProducer producer;
// TextMessage message;
//构造ConnectionFactory实例对象,此处采用ActiveMq的实现jar
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,
"tcp://10.20.8.198:61616");
try {
//构造从工厂得到连接对象
connection = connectionFactory.createConnection();
//启动
connection.start();
//获取操作连接
session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
//获取session注意参数值FirstTopic是一个服务器的topic(与queue消息的发送相比,这里是唯一的不同)
destination = session.createTopic("FirstTopic");
//得到消息生成者【发送者】
producer = session.createProducer(destination);
//设置不持久化,此处学习,实际根据项目决定
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
//构造消息,此处写死,项目就是参数,或者方法获取
sendMessage(session, producer);
session.commit();
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
}
}
}
}
3.2 启动多个客户端监听来接收topic的消息:
publicclass ReceiveTopicimplements Runnable {
private StringthreadName;
ReceiveTopic(String threadName) {
this.threadName = threadName;
}
publicvoid run() {
// ConnectionFactory:连接工厂,JMS用它创建连接
ConnectionFactory connectionFactory;
// Connection:JMS客户端到JMS Provider的连接
Connection connection =null;
// Session:一个发送或接收消息的线程
Session session;
// Destination:消息的目的地;消息发送给谁.
Destination destination;
//消费者,消息接收者
MessageConsumer consumer;
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,"tcp://10.20.8.198:61616");
try {
//构造从工厂得到连接对象
connection = connectionFactory.createConnection();
//启动
connection.start();
//获取操作连接,默认自动向服务器发送接收成功的响应
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//获取session注意参数值FirstTopic是一个服务器的topic
destination = session.createTopic("FirstTopic");
consumer = session.createConsumer(destination);
while (true) {
//设置接收者接收消息的时间,为了便于测试,这里设定为100s
TextMessage message = (TextMessage) consumer
.receive(100 * 1000);
if (null != message) {
System.out.println("线程"+threadName+"收到消息:" + message.getText());
} else {
continue;
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
}
}
}
publicstaticvoid main(String[] args) {
//这里启动3个线程来监听FirstTopic的消息,与queue的方式不一样三个线程都能收到同样的消息
ReceiveTopic receive1=new ReceiveTopic("thread1");
ReceiveTopic receive2=new ReceiveTopic("thread2");
ReceiveTopic receive3=new ReceiveTopic("thread3");
Thread thread1=new Thread(receive1);
Thread thread2=new Thread(receive2);
Thread thread3=new Thread(receive3);
thread1.start();
thread2.start();
thread3.start();
}
}
4 queue方式的消息处理示例
参考上一期文章:开源jms服务ActiveMQ的负载均衡+高可用部署方案探索。
相关推荐
Apache ActiveMQ Queue Topic 详解 教程 加入代码解释说明
一个jms activemq Topic 消息实例 关于jms JMS 是接口,相当于jdbc ,要真正使用它需要某些厂商进行实现 ,即jms provider 常见的jms provider 有 ActiveMQ JBoss 社区所研发的 HornetQ (在jboss6 中默认即可以...
Queue与Topic的比较,学习JMS和activemq必须看的资料之一
springboot2整合activemq的demo内含queue消息和topic消息,需要使用者修改application.yml的连接地址信息就可以了,端口号是默认的可以不用修改
主要介绍了详解Springboot整合ActiveMQ(Queue和Topic两种模式),文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
本人在学习activemq,然后 测试完成的demo, 包含了queue,topic,持久化到mysql,订阅模式,包好用
文章目录ActiveMQ简介1、ActiveMQ简介2、ActiveMQ下载SpringBoot整合ActiveMQ1、新建SpringBoot项目2、项目结构3、相关配置信息4、ActiveMQ配置类Queue队列模式1、队列生产者2、队列消费者3、测试效果Topic模式1、...
activemq queue模式,事务、应答、转发模式、MessageConsumer的receive阻塞方法的测试
ActiveMQ的两种消息模式的例子,queue和topic,注释很详细,告诉你每步做什么,非springboot整合
百度spring整合activemq 发现几乎都只是在xml文件配置固定的消息队列而且太麻烦。并没有根据需求进行动态生成主题和队列。本文档参考了纯粹的activemq java代码和百度上的demo,很简洁的实现了动态消息队列的生成和...
ActiveMQ队列消息过期时间设置和自动清除解决方案.docx
欢迎下载ActiveMQ Topic 实例!
activemq 虚拟topic配置,可以将一个 topic转发为多个队列和多个topic或者将一个队列转发为多个topic和多个队列
springboot整合activemq案例,queue,topic两种模式 定时运行和controller请求运行两种方式
ActiveMQ是Apache下的开源项目,完全支持JMS1.1和J2EE1.4规范的JMS Provider实现。... JMS中定义了两种消息模型: 点对点(point to point, queue) 发布/订阅(publish/subscribe,topic)。
activeMQ JMS 3种创建方式 公共方式 QUEUE TOPIC
activemq的queue队列模式的maven,spring的demo
详细描述了ActiveMQ消息过期-时间设置和自动清除解决方案。
ActiveMQ高并发处理方案ActiveMQ高并发处理方案 超级字数补丁超级字数补丁
ActiveMQ-Queues点对点消息-Receive+Listener方式:参考博文:http://blog.csdn.net/ABAP_Brave/article/details/53443725