上一期介绍了我们项目要用到activemq来作为jms总线，并且给大家介绍了activemq的集群和高可用部署方案，本期给大家再介绍下，如何根据自己的项目需求，更好地使用activemq的两种消息处理模式。
对比项
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则是以点对点的形式通知多个处于监听状态的客户端中的一个。
通过增加监听客户端的并发数来验证，topic的消息推送，是否会因为监听客户端的并发上升而出现明显的下降，测试环境的服务器为ci环境的activemq，客户端为我的本机。
从实测的结果来看，topic方式发送的消息，发送和接收的效率，在一个订阅者和100个订阅者的前提下没有明显差异，但在500个订阅者（线程）并发的前提下，效率差异很明显（由于500线程并发的情况下，我本机的cpu占用率已高达70-90%，所以无法确认是我本机测试造成的性能瓶颈还是topic消息发送方式存在性能瓶颈，造成效率下降如此明显）。
topic方式发送的消息与queue方式发送的消息，发送和接收的效率，在一个订阅者和100个订阅者的前提下没有明显差异，但在500个订阅者并发的前提下，topic方式的效率明显低于queue。
queue方式发送的消息，在一个订阅者、100个订阅者和500个订阅者的前提下，发送和接收的效率没有明显变化。
topic实测数据：
发送者发送的消息总数
所有订阅者接收到消息的总数
消息发送和接收平均耗时
单订阅者
100
101ms
100订阅者
10000
103ms
500订阅者
50000
14162ms
queue实测数据：
96ms
100ms
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();
参考上一期文章：开源jms服务activemq的负载均衡+高可用部署方案探索。
![Java小案例——随机数猜测随机数猜测[图]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)