【中间件】ActiveMQ的queue以及topic消息处理机制浅析

• queue与topic的技术特点对比

Topic和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实测数据：

Queue实测数据：

• topic方式的消息处理示例

通过客户端代码调用来发送一个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 {

privatestaticfinalint SEND_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) {

}

}

}

}

启动多个客户端监听来接收topic的消息：

publicclass ReceiveTopic implements 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();

}

}