`
小码哥BASE64
  • 浏览: 124328 次
社区版块
存档分类
最新评论

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发送消息到目标Queuereceiver可以异步接收这个Queue上的消息。Queue上的消息如果暂时没有receiver来取,也不会丢失。

消息发布接收策略

一对多的消息发布接收策略,监听同一个topic地址的多个sub都能收到publisher发送的消息。Sub接收完通知mq服务器

一对一的消息发布接收策略,一个sender发送的消息,只能有一个receiver接收。receiver接收完后,通知mq服务器已接收,mq服务器对queue里的消息采取删除或其他操作。

          Topicqueue的最大区别在于topic是以广播的形式,通知所有在线监听的客户端有新的消息,没有监听的客户端将收不到消息;而queue则是以点对点的形式通知多个处于监听状态的客户端中的一个。

 

2    topicqueue方式的消息处理效率比较

        通过增加监听客户端的并发数来验证,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;

        // ConnectionJMS客户端到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;

           // ConnectionJMS客户端到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的负载均衡+高可用部署方案探索

 

分享到:
评论
2 楼 dirkmurphy 2016-02-18  
哦 我知道了... 免回
1 楼 dirkmurphy 2016-02-18  
Queue方式 不是point to point的吗?怎么还有100个订阅者?

相关推荐

    ActiveMQ的queue和topic两种模式的示例演示参照.pdf

    在ActiveMQ中,有两种主要的消息传递模式:Queue和Topic。这两种模式都是基于JMS(Java Message Service)标准,用于在分布式环境中实现异步通信。理解它们的区别和应用场景至关重要。 **Queue(队列)模式** ...

    消息队列 Queue与Topic区别.docx

    这两种模型的核心区别在于消息如何被传递以及如何被接收。 #### 二、点对点(Point-to-Point, PTP) 点对点消息传递模型是基于队列(Queue)的消息传递方式,其特点是: - **单向性**:一个消息生产者(Producer...

    Queue与Topic的比较

    Queue 和 Topic 是 JMS(Java Message Service)中两种基本的消息模式,分别对应 Point-to-Point 和 Publish/Subscribe 模式。 Queue 模式 在 Queue 模式中,一条消息仅能被一个消费者(Consumer)接收。如果在...

    activemq 虚拟topic与路由功能

    除了虚拟Topic外,ActiveMQ还支持另一种类似但功能略有不同的概念——复合队列(Composite Queue)。两者的主要区别在于: - **虚拟Topic**:主要用于将一个Topic的消息转发到多个目的地(Topic或队列),适合于...

    详解Springboot整合ActiveMQ(Queue和Topic两种模式)

    在本文中,我们将深入探讨如何将SpringBoot与Apache ActiveMQ集成,特别关注Queue和Topic两种消息传递模式。首先,我们来了解一下ActiveMQ的基本概念和功能。 Apache ActiveMQ是一款开源的消息中间件,由Apache软件...

    Springboot整合ActiveMQ(Queue和Topic两种模式)

    通过`@Value`注入队列和主题的名称,并分别定义`ActiveMQQueue`和`ActiveMQTopic`的@Bean,如以下示例所示: ```java @Component @EnableJms public class ConfigBean { @Value("${myqueue}") private String ...

    ActiveMQ Topic 实例

    在分布式系统中,消息队列(Message Queue)作为一种中间件,起到了解耦、异步处理以及负载均衡的作用。Apache ActiveMQ 是一个非常流行的开源消息代理,它实现了多种消息协议,如OpenWire、STOMP、AMQP和MQTT等。本...

    ActiveMQ的队列、topic模式

    本文将深入探讨ActiveMQ中的两种主要消息模式:队列(Queue)和主题(Topic)。 1. **队列(Queue)模式**: 队列模式遵循“发布/订阅”模型,但是一对一的。每个消息只能被一个消费者接收并处理。当一个消息被...

    JMS中topic和queue两种实现方式

    JMS主要提供了两种消息模式:主题(Topic)和队列(Queue),这两种模式在实现方式和功能上有所不同。 1. 主题(Topic): - **发布/订阅模型**:主题基于发布/订阅模型,其中多个生产者可以发布消息到一个特定的...

    Spring+ActiveMQ消息队列+前台接收消息

    4. **消息类型**:JMS支持两种消息模型:点对点(Queue)和发布/订阅(Topic)。Queue模式下,消息会被一个消费者接收并删除;Topic模式下,消息可以被多个订阅者接收。 5. **消息的持久化**:ActiveMQ允许配置消息...

    activeMQ收发工具.rar

    2. **消息模型**:掌握点对点(Queue)和发布/订阅(Topic)两种消息模型。点对点模型确保每个消息仅被一个消费者接收,而发布/订阅模型允许多个订阅者接收相同的消息。 3. **消息类型**:理解JMS提供的不同消息...

    Springboot-activeMQ

    ActiveMQ是Apache下的开源项目,完全支持JMS1.1和J2EE1.4规范的JMS Provider实现。... JMS中定义了两种消息模型:  点对点(point to point, queue)  发布/订阅(publish/subscribe,topic)。

    Activemq同时支持多个Topic类型通信,并且配置添加到服务里面方便管理

    此外,ActiveMQ提供了Web管理界面,可以通过浏览器访问`http://localhost:8161/admin`,方便地管理和监控Topic、Queue以及其他配置。 下载的`apache-activemq-5.14.5`压缩包包含了完整的ActiveMQ服务器及其相关文档...

    Apache ActiveMQ教程 JMS 整合Tomcat

    启动ActiveMQ服务器有两种常见方式:一是通过bin目录下的activemq.bat脚本直接启动;二是在JVM环境中嵌套启动,利用ant工具进行,具体步骤为导航至example目录,执行ant embedBroker命令。此外,ActiveMQ还提供了一...

    Apache_ActiveMQ教程

    启动方法有两种:一种是直接运行bin目录下的activemq.bat脚本,另一种是通过在JVM中嵌套启动。部署完成后,可以通过ActiveMQ消息管理后台系统进行管理和监控,该系统可以通过浏览器访问地址***。 ActiveMQ支持Point...

    MessageQueue API ActiveMQ Core 5_2_0_0-fuse API

    4. **目的地**:消息的接收方,分为队列(Queue)和主题(Topic)两种类型。队列遵循点对点模型,每个消息仅由一个消费者接收;主题遵循发布/订阅模型,多个订阅者可以接收到同一消息。 5. **连接**(Connection):...

    C#,activemq,mq

    为了确保高可用性和可靠性,ActiveMQ支持主题(Topics)和队列(Queues)两种模型。主题采用发布/订阅模式,多订阅者可以同时接收相同的消息;而队列遵循先进先出(FIFO)原则,每个消息仅被一个消费者接收。此外,...

    activemq_spring.rar_Spring和ActiveMQ_spring_消息中间件_消息发布订阅_消息订阅

    6. **实现消息的发布订阅模式**:在ActiveMQ中,有两种消息传递模式:点对点(Queue)和发布订阅(Topic)。点对点模式下,每个消息只有一个消费者;发布订阅模式下,一个消息可以被多个消费者(订阅者)接收。在...

    springboot整合activemq案例

    在本案例中,我们将关注如何在Spring Boot应用中集成ActiveMQ,以及如何使用两种主要的消息类型——Queue和Topic。Queue是一种点对点的消息模型,每个消息仅被一个消费者接收;而Topic则遵循发布/订阅模型,一个消息...

    ActiveMQ接受和发送工具.rar

    3. **消息模型**:ActiveMQ支持两种消息模型——点对点(Queue)和发布/订阅(Topic)。点对点模型适用于一对一通信,而发布/订阅模型适用于一对多广播式通信。 4. **协议支持**:ActiveMQ不仅支持JMS,还支持AMQP...

Global site tag (gtag.js) - Google Analytics