ActiveMQ的queue以及topic两种消息处理机制分析

布满荆棘的人生 2022-06-13 02:57 229阅读 0赞

上一期介绍了我们项目要用到activeMQ来作为jms总线，并且给大家介绍了activeMQ的集群和高可用部署方案，本期给大家再介绍下，如何根据自己的项目需求，更好地使用activeMQ的两种消息处理模式。

##### 1    queue与topic的技术特点对比 #####

<table style="color:rgb(0,0,0); font-family:Arial,Helvetica,San-serif; font-size:14px"> 
 <tbody> 
 <tr> 
 <td> Topic </td> 
 <td> Queue </td> 
 </tr> 
 <tr> 
 <td> 概要 </td> 
 <td> Publish Subscribe messaging 发布订阅消息 </td> 
 <td> Point-to-Point 点对点 </td> 
 </tr> 
 <tr> 
 <td> 有无状态 </td> 
 <td> topic数据默认不落地，是无状态的。 </td> 
 <td> Queue数据默认会在mq服务器上以文件形式保存，比如Active MQ一般保存在$AMQ_HOME\data\kr-store\data下面。也可以配置成DB存储。 </td> 
 </tr> 
 <tr> 
 <td> 完整性保障 </td> 
 <td> 并不保证publisher发布的每条数据，Subscriber都能接受到。 </td> 
 <td> Queue保证每条数据都能被receiver接收。 </td> 
 </tr> 
 <tr> 
 <td> 消息是否会丢失 </td> 
 <td> 一般来说publisher发布消息到某一个topic时，只有正在监听该topic地址的sub能够接收到消息；如果没有sub在监听，该topic就丢失了。 </td> 
 <td> Sender发送消息到目标Queue，receiver可以异步接收这个Queue上的消息。Queue上的消息如果暂时没有receiver来取，也不会丢失。 </td> 
 </tr> 
 <tr> 
 <td> 消息发布接收策略 </td> 
 <td> 一对多的消息发布接收策略，监听同一个topic地址的多个sub都能收到publisher发送的消息。Sub接收完通知mq服务器 </td> 
 <td> 一对一的消息发布接收策略，一个sender发送的消息，只能有一个receiver接收。receiver接收完后，通知mq服务器已接收，mq服务器对queue里的消息采取删除或其他操作。 </td> 
 </tr> 
 </tbody> 
</table>

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

<table style="color:rgb(0,0,0); font-family:Arial,Helvetica,San-serif; font-size:14px"> 
 <tbody> 
 <tr> 
 <td> 　 </td> 
 <td> 发送者发送的消息总数 </td> 
 <td> 所有订阅者接收到消息的总数 </td> 
 <td> 消息发送和接收平均耗时 </td> 
 </tr> 
 <tr> 
 <td> 单订阅者 </td> 
 <td> 100 </td> 
 <td> 100 </td> 
 <td> 101ms </td> 
 </tr> 
 <tr> 
 <td> 100订阅者 </td> 
 <td> 100 </td> 
 <td> 10000 </td> 
 <td> 103ms </td> 
 </tr> 
 <tr> 
 <td> 500订阅者 </td> 
 <td> 100 </td> 
 <td> 50000 </td> 
 <td> 14162ms </td> 
 </tr> 
 </tbody> 
</table>

Queue实测数据：

<table style="color:rgb(0,0,0); font-family:Arial,Helvetica,San-serif; font-size:14px"> 
 <tbody> 
 <tr> 
 <td> 　 </td> 
 <td> 发送者发送的消息总数 </td> 
 <td> 所有订阅者接收到消息的总数 </td> 
 <td> 消息发送和接收平均耗时 </td> 
 </tr> 
 <tr> 
 <td> 单订阅者 </td> 
 <td> 100 </td> 
 <td> 100 </td> 
 <td> 96ms </td> 
 </tr> 
 <tr> 
 <td> 100订阅者 </td> 
 <td> 100 </td> 
 <td> 100 </td> 
 <td> 96ms </td> 
 </tr> 
 <tr> 
 <td> 500订阅者 </td> 
 <td> 100 </td> 
 <td> 100 </td> 
 <td> 100ms </td> 
 </tr> 
 </tbody> 
</table>

##### 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 \{

**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) \{

###### 3.2     启动多个客户端监听来接收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();

##### 4     queue方式的消息处理示例 #####

参考上一期文章：开源jms服务ActiveMQ的负载均衡+高可用部署方案探索。

欢迎访问我们的技术交流群425783133

发表评论取消回复

表情：

评论列表（有 0 条评论，229人围观）

ActiveMQ的queue以及topic两种消息处理机制分析

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 springboot整合Activemq,queue和topic两种消息模式

相关消息队列MQ实践----实现Queue(队列消息)和Topic(主题消息)两种模式

相关 ActiveMQ消息传送机制以及ACK机制详解

相关 ActiveMQ的queue以及topic两种消息处理机制分析

相关【MQ】ActiveMQ（二）——消息处理机制

相关 ActiveMQ发送queue消息

相关 ActiveMQ定时删除过期Topic和Queue

相关 ActiveMQ的两种消息形式。

相关 ActiveMQ消息传送机制以及ACK机制详解

相关 ActiveMQ定时删除过期的Topic和Queue