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

        上一期介绍了我们项目要用到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的负载均衡+高可用部署方案探索。

 

时间： 2024-08-30 01:42:42

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

Ajax中解析Json的两种方法对比分析

  这里给大家介绍的是Ajax中解析Json的两种方法对比分析,十分的实用,本文为学习笔记,属新手文章,欢迎指教! eval(); //此方法不推荐 JSON.parse(); //推荐方法 一.两种方法的区别 我们先初始化一个json格式的对象: ? 1 2 3 4 5 var jsonDate = '{ "name":"周星驰","age":23 }'   var jsonObj = eval( '(' + jsonDate + ')' );

ReentrantLock和synchronized两种锁定机制

ReentrantLock和synchronized两种锁定机制 应用synchronized同步锁 把代码块声明为 synchronized,使得该代码具有 原子性(atomicity)和 可见性(visibility). 原子性意味着一个线程一次只能执行由一个指定监控对象(lock)保护的代码,从而防止多个线程在更新共享状态时相互冲突. 可见性类似volatile关键字. 应用ReentrantLock显示锁 ReentrantLock 类实现了 Lock ,它拥有与 synchronize

SEO还是SEM 两种网络营销方式分析

有人将所有通过网络途径达到营销目的的方式统称网络营销,有人分不清楚SEM与SEO的区别,有人甚至不知道有微信营销.微博营销等营销方式.当然,这并没有错,毕竟,隔行如隔山.但,如果你准备通过网络途径拓展你的营销道路,一些基本的概念.作用.区别.优势与不足这些还是需要懂的.目前微博营销已经成为过去式,微信营销还没有完全形成气候,平台营销门槛高,束缚也多.新的营销方式还没有得到完善,而搜索引擎强大的粘附力仍然存在.因而,SEM与SEO仍然是目前主流的网络营销方式,广州网站建设奇亿网络小编今天就来跟大家

J2ME网络编程两种方法的分析

本文描述了在J2me中开发主要使用的网络连接方法,分别详细介绍了使用http和socket两种方法. HttpConnection 首先我们先来看一个简单的例子吧: 主要用到的java包: javax.microedition.io.*; public String requestGET(String URLString,String URL) throws IOException{ // =====================================================

java两种单例模式用法分析_java

本文实例讲述了java两种单例模式用法.分享给大家供大家参考,具体如下: 按照加载方式的不同,单例模式有两种实现: private:只能在同一个类中使用 static:该类是类方法,不能调用实例方法./类全局变量 final:方法或成员变量不能被修饰 1.饿汉式 public class EagerSigleton{ private static final EagerSigleton instance=new EagerSigleton(); private EagerSigleton(){}

动态加载jQuery的两种方法实例分析_jquery

本文实例讲述了动态加载jQuery的两种方法.分享给大家供大家参考.具体如下: 第一种方法参考本站之前有人发的代码,增加了加载检测: 第二种方法来自去年的12306刷票脚本. 第一种方法: function withjQuery(callback) { if(!(window.jQuery)) { var js = document.createElement('script'); js.setAttribute('src', 'https://dynamic.12306.cn/otsweb/j

Ajax中解析Json的两种方法对比分析_json

eval();  //此方法不推荐 JSON.parse();  //推荐方法 一.两种方法的区别 我们先初始化一个json格式的对象: var jsonDate = '{ "name":"周星驰","age":23 }' var jsonObj = eval( '(' + jsonDate + ')' ); // eval();方法 var jsonObj = JSON.parse( jsonDate ); // JSON.parse(); 方

Hadoop MapReduce最常见的两种容错场景分析

本文将分析Hadoop MapReduce(包括MRv1和MRv2)的两种常见的容错场景,第一种是,作业的某个任务阻塞了,长时间占用资源不释放,如何处理?另外一种是,作 业的Map http://www.aliyun.com/zixun/aggregation/17034.html">Task全部运行完成后,在Reduce Task运行过程中,某个Map Task所在节点挂了,或者某个Map Task结果存放磁盘损坏了,该如何处理? 第一种场景:作业的某个任务阻塞了,长时间占用资源不释放,

Java中的ReentrantLock和synchronized两种锁机制的对比

原文:http://www.ibm.com/developerworks/cn/java/j-jtp10264/index.html 多线程和并发性并不是什么新内容,但是 Java 语言设计中的创新之一就是,它是第一个直接把跨平台线程模型和正规的内存模型集成到语言中的主流语言.核心类库包含一个 Thread 类,可以用它来构建.启动和操纵线程,Java 语言包括了跨线程传达并发性约束的构造 -- synchronized 和 volatile.在简化与平台无关的并发类的开发的同时,它决没有使并发