`
guomingzhang2008
  • 浏览: 160777 次
  • 性别: Icon_minigender_1
  • 来自: 上海
社区版块
存档分类
最新评论

ActiveMQ

    博客分类:
  • jms
阅读更多
2. JMS Provider(ActiveMQ)

特性及优势

1. 实现JMS1.1规范,支持J2EE1.4以上。

2. 可运行与任何JVM和大部分web容器(ActiveMQ works great in any JVM)

3. 支持多种语言客户端(java, C, C++, Ajax, ActionScript等等)

4. 支持多种协议(stomp, openwire, REST)

5. 良好的Spring支持(ActiveMQ has great Spring Support)

6. 速度很快,JBossMQ的十倍(ActiveMQ is very fast; often 10x faster than JBossMQ)

7. 与OpenJMS、JBossMQ等开源jms provider相比,ActiveMQ有apache的支持,持续发展的优势明显



Queue与Topic的比较

1. JMS Queue执行load balancer语义

    一条消息仅能被一个consumer收到。如果在message发送的时候没有可用的consumer,那么它讲被保存一直到能处理该message的consumer可用。如果一个consumer收到一条message后却不响应它,那么这条消息将被转到另外一个consumer那儿。一个Queue可以有很多consumer,并且在多个可用的consumer中负载均衡。

2. Topic实现publish和subscribe语义

    一条消息被publish时,他将发送给所有感兴趣的订阅者,所以零到多个subscriber将接收到消息的一个拷贝。但是在消息代理接收到消息时,只有激活订阅的subscriber能够获得消息的一个拷贝。

3. 分别对应两种消息模式

    Point-to-Point(点对点),Publisher/Subscriber Model(发布/订阅者)

    其中在Publicher/Subscriber模式下又有Nondurable subscription(非持久化订阅)和durable subscription(持久化订阅)两种消息处理方式。

Point-to-Point(点对点)消息模式开发流程

1. 生产者(producer)开发流程
    1.1 创建 Connection
Java代码

// 根据url,user和password创建一个jms Connection。  
    ActiveMQConnectionFactory connectionFactory   =   new ActiveMQConnectionFactory (user, password, url);  
    connection = connectionFactory.createConnection();  
    connection.start();  


    1.2 创建Session
Java代码
    /**在connection的基础上创建一个session,同时设置是否支持事务ACKNOWLEDGE标识。 
    • AUTO_ACKNOWLEDGE:自动确认模式。一旦接收方应用程序的方法调用从处理消息处返回,会话对象就会确认消息的接收。  
    • CLIENT_ACKNOWLEDGE:客户端确认模式。会话对象依赖于应用程序对被接收的消息调用一个acknowledge()方法。一旦这个方法被调用,会话会确认最后一次确认之后所有接收到的消息。这种模式允许应用程序以一个调用来接收,处理并确认一批消息。注意:在管理控制台中,如果连接工厂的Acknowledge Policy(确认方针)属性被设置为"Previous"(提前),但是你希望为一个给定的会话确认所有接收到的消息,那么就用最后一条消息来调用acknowledge()方法。  
    • DUPS_OK_ACKNOWLEDGE:允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回,会话对象就会确认消息的接收;而且允许重复确认。在需要考虑资源使用时,这种模式非常有效。注意:如果你的应用程序无法处理重复的消息的话,你应该避免使用这种模式。如果发送消息的初始化尝试失败,那么重复的消息可以被重新发送。 
    •  SESSION_TRANSACTED**/  
    Session session = connection.createSession(  
    transacted, Session.AUTO_ACKNOWLEDGE);  


    1.3 创建Destination对象
Java代码
    //需指定其对应的主题(subject)名称,producer和consumer将根据subject来发送/接收对应的消息  
    if (topic) {  
        destination = session.createTopic(subject);  
    } else {  
        destination = session.createQueue(subject);  
    }  


    1.4 创建MessageProducer
Java代码
    //根据Destination创建MessageProducer对象,同时设置其持久模式。   
    MessageProducer producer = session.createProducer(destination);  
    if (persistent) {  
          producer.setDeliveryMode(DeliveryMode.PERSISTENT);  
    } else {  
          producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);  
    }  
    if (timeToLive != 0) {  
           producer.setTimeToLive(timeToLive);  
    }   



    1.5 发送消息到队列(Queue)
Java代码
    //封装TextMessage消息,使用MessageProducer的send方法将消息发送出去。  
    TextMessage message = session.createTextMessage("createMessageText");  
    producer.send(message);  


2. 消费者(consumer)开发流程
    2.1 实现MessageListener接口
Java代码 
    //消费者类必须实现MessageListener接口,然后在onMessage方法中监听消息到达处理。  



    2.2 创建Connection
Java代码 
    //根据url,user和password创建一个jms Connection,如果是durable模式,还需要给connection设置一个clientId。  
    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);  
    Connection connection = connectionFactory.createConnection();  
    if (durable && clientId != null && clientId.length() > 0 && !"null".equals(clientId)) {  
          connection.setClientID(clientId);  
    }  
    connection.setExceptionListener(this);  
    connection.start();  


    2.3 创建Session和Destination
Java代码  
    //与产品类似  



    2.4 创建replayProducer【可选】
Java代码  
    //可以用来将消息处理结果发送给producer。   
    replyProducer = session.createProducer(null);  
    replyProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT)




    2.5 创建MessageConsumer
Java代码  
    //根据Destination创建MessageConsumer对象。  
    MessageConsumer consumer = null;  
    if (durable && topic) {  
         consumer = session.createDurableSubscriber((Topic)destination, consumerName);  
    } else {  
         consumer = session.createConsumer(destination);  
    }


    2.6 消费message
Java代码 
    //在onMessage()方法中接收producer发送过来的消息进行处理,并可以通过replyProducer反馈信息给producer   
    if (message.getJMSReplyTo() != null) {   
        replyProducer.send(message.getJMSReplyTo(),   
        session.createTextMessage("Reply: " + message.getJMSMessageID()));   
    }    



Publish/Subscriber(发布/订阅者)消息开发模式
1. 订阅者(Subscriber)开发流程
    1.1 实现MessageListener接口
  在onMessage()方法中监听发布者发出的消息队列,并做相应处理。

    1.2 创建Connection
          根据url, user和password创建一个jms connection

    1.3 创建Session
  在connection的基础上创建一个session,同时设置是否支持和ACKNOWLEDGE标志。

    1.4 创建 Topic
          创建两个Topic,topictest.message用于接收发布者发出的消息,topictest.control用于向发布者发送消息,实现双方的交互。

    1.5 创建consumer和producer对象
  根据topictest.message创建consumer,根据topictest.control创建producer

    1.6 接收处理消息
  在onMessage()方法中,对收到的消息进行处理,可直接简单在本地显示消息,或者根据消息内容不同处理对应的业务逻辑(比如:数据库更新、文件操作等等),并且可以使用   producer对象处理结果返回给发布者。

2. 发布者(Publisher)开发流程
     2.1 实现MessageListener接口
   在onMessage()方法中接收订阅者的反馈消息。

     2.2 创建Connection
   根据url, user和password 创建一个 jms Connection。

     2.3 创建session
   在connection的基础上创建一个session,同时设置是否支持事务和ACKNOWLEDGE标志。

     2.4 创建Topic
   创建两个Topic,topictest.messages用于向订阅者发布消息,topictest.control用于接收订阅者反馈的消息。这两个Topic与订阅者开发流程中的topic是一一对应的。

     2.5 创建consumer和producer对象
   根据topictest.message创建publisher;
   根据topictest.control穿件consumer,同时监听订阅者反馈的消息。


3. JMS For Spring
      Spring提供了用于简化JMS API使用的抽象框架,并且对用户屏蔽了JMS API中1.0.2和1.1版本的差异。

      JMS的功能大致上分为两块,叫做消息制造和消息消耗。JmsTemplate用于制造消息和同步消息接收。和J2EE的事件驱动Bean风格类似,对于异步接收消息,Spring提供了一些消息监听容器来创建消息驱动的POJO(MDP)。

1. Spring 配置 connectionFactory
Xml代码 
    <!--客户端使用普通传输方式:tcp://localhost:61616   
        此处需加以注意的是Listener端的borkerURL使用了failover传输方式:   
        failover:(tcp://localhost:61616)?initialReconnectDelay=100&amp;  
    maxReconnectAttempts=5   
        failover transport是一种重新连接机制,用于建立可靠的传输。此处配置的是一旦ActiveMQ broker中断,Listener端将每隔100ms自动尝试连接,直至成功连接或重试5次连接失败为止。  
        failover还支持多个borker同时提供服务,实现负载均衡的同时可增加系统容错性,格式: failover:(uri1,...,uriN)?transportOptions  
    -->  
    <bean id="jmsFactory" destroy-method="stop" class="org.apache.activemq.pool.PooledConnectionFactory">  
            <property name="connectionFactory">  
                <bean class="org.apache.activemq.ActiveMQConnectionFactory">  
                    <property name="brokerURL" value="tcp://localhost:61616" />  
                     <property name="userName" value="${activemq.username}" />  
                    <property name="password" value="${activemq.password}" />  
                </bean>  
            </property>  
    </bean>  
 

2. Spring JmsTemplate
Xml代码 
    <!-- Spring JMS Template  
            JmsTemplate 类的实例 一经配置便是线程安全 的。 要清楚一点,JmsTemplate   
            是有状态的,因为它维护了 ConnectionFactory 的引用,但这个状态时不是会话状态。  
    -->  
        <bean id="myJmsTemplate"  
            class="org.springframework.jms.core.JmsTemplate">  
            <property name="connectionFactory" ref="jmsFactory" />  
            <property name="defaultDestinationName" value="MySubject" />  
            <!--JMS API有两种发送方法,一种采用发送模式、  
    优先级和存活时间作为服务质量(QOS)参数,  
                默认{deliveryMode:2(1),priority:4,timeToLive:0}  
                另一种使用无需QOS参数的缺省值方法。  
                <property name="explicitQosEnabled" value="true"/>  
                <property name="deliveryMode" value="2"/>  
                <property name="priority" value="4"/>  
                            <property name="timeToLive" value="1000"/>  
             -->  
            <!-- 
                <property name="receiveTimeout" value="1000" /> 
             -->  
        </bean>  
 

3. 发送的接收消息
Xml代码  
    <bean id="destinationTopic"  
            class="org.apache.activemq.command.ActiveMQTopic">  
            <constructor-arg index="0" value="HelloWorldTopic" />  
        </bean>  
        <!-- 读取信息 -->  
        <bean id="consumer" class="com.d1xn.jms.demo.Consumer">  
            <property name="jmsTemplate" ref="myJmsTemplate" />  
        </bean>  
      
        <!-- 发送信息 -->  
        <bean id="producerTopic" class="com.d1xn.jms.demo.ProducerTopic">  
            <property name="jmsTemplet" ref="myJmsTemplate" />  
            <property name="destination" ref="destinationTopic" />  
        </bean>  
        <!-- 消息监听 -->  
      
        <bean id="listenerContainerTopic"  
    class="org.springframework.jms.listener.DefaultMessageListenerContainer">  
            <property name="connectionFactory" ref="jmsFactory" />  
            <property name="destination" ref="destinationTopic" />  
            <property name="messageListener" ref="consumer" />  
             <!—持久化客户端ID -->  
            <property name="clientId" value="clientId_001" />  
            <property name="subscriptionDurable" value="true" />  
            <property name="durableSubscriptionName" value="My_001" />  
    </bean>  


说明(基于ActiveMQ5.4.2版本):
        1、Web Console 的安全配置可参考  
    将conf/jetty.xml下面一段xml配置:
Xml代码 
    <bean id="securityConstraint" class="org.eclipse.jetty.http.security.Constraint">  
        <property name="name" value="BASIC" />                                         
        <property name="roles" value="admin" />                                        
        <property name="authenticate" value="true" />                                  
    </bean>                                                                          
 

   authenticate值设置为true即可!那用户名/密码的配置是在conf/jetty-realm.properties!

   详细可参考http://activemq.apache.org/web-console.html


2、连接安全配置可参考
     将conf/activemq-security.xml中如下的配置
Xml代码  
    <plugins>                                                                                   
     <!-- Configure authentication; Username, passwords and groups -->                               
     <simpleAuthenticationPlugin anonymousAccessAllowed="false">                                    
        <users>                                                                                      
            <authenticationUser username="system" password="${activemq.password}"                    
                groups="admins"/>                                                                    
            <!--<authenticationUser username="user" password="${guest.password}"                     
                groups="users"/>                                                                     
            <authenticationUser username="guest" password="${guest.password}" groups="guests"/>-->   
        </users>                                                                                     
     </simpleAuthenticationPlugin>                                                                   
    </plugins>                                                                       
               

            copy至conf/activemq.xml中
Xml代码  
    <persistenceAdapter>                                    
        <kahaDB directory="${activemq.base}/data/kahadb"/>  
    </persistenceAdapter>                                   

    的下面(这是简单的用户名、密码的认证方式)!

   用户名、密码的可在conf/credentials.properties配置!

   详细可参考:

     http://activemq.apache.org/security.html

     http://activemq.apache.org/xml-reference.html


本文转载于:http://boy00fly.iteye.com/blog/1103586
分享到:
评论

相关推荐

    activeMQ收发工具.rar

    ActiveMQ是中国最流行的开源消息中间件之一,由Apache软件基金会开发。它基于Java Message Service (JMS) 规范,提供了可靠的消息传递功能,适用于分布式系统中的应用间通信。本压缩包“activeMQ收发工具.rar”包含...

    ActiveMQ高并发处理方案

    ### ActiveMQ高并发处理方案详解 #### 一、引言 在现代分布式系统中,消息队列作为异步通信的核心组件之一,对于提高系统的吞吐量、降低响应时间和实现服务解耦等方面起着至关重要的作用。Apache ActiveMQ作为一款...

    apache-activemq-5.17.3

    Apache ActiveMQ是开源的、基于Java消息服务(JMS)的应用服务器,它是Apache软件基金会的一部分。这个名为"apache-activemq-5.17.3"的压缩包包含了ActiveMQ的5.17.3版本,这是一个稳定且功能丰富的发布版本。在深入...

    apache-activemq-5.16.5

    Apache ActiveMQ是业界广泛使用的开源消息中间件,它基于Java消息服务(JMS)标准,提供了高度可扩展、可靠的异步通信能力。标题"apache-activemq-5.16.5"指的是该软件的一个特定版本,即5.16.5版本,通常每个新版本...

    Jmeter测试ActiveMQ性能报告

    本报告详细阐述了使用JMeter对ActiveMQ进行性能测试的过程和结果,旨在评估ActiveMQ在JMS(Java消息服务)环境下的性能表现。JMeter作为一个强大的负载和性能测试工具,被广泛用于测试各种应用程序,包括消息中间件...

    jmx监控activeMQ监控

    jmx监控ActiveMQ监控 jmx(Java Management Extensions)是一种Java技术,为Java应用程序提供了管理和监控的功能。ActiveMQ是Apache软件基金会下的一个开源消息队列系统,提供了高效、可靠的消息传递服务。在生产...

    Linux下activeMQ的启动和停止.docx

    在Linux环境下,Apache ActiveMQ是一个广泛使用的开源消息代理和队列服务器,它是Java Message Service (JMS) 的实现,能够处理大量的并发消息传递。ActiveMQ提供了高可用性、可扩展性和稳定性,使得它成为分布式...

    apache-activemq-5.15.9.rar

    Apache ActiveMQ是Apache软件基金会开发的一个开源消息中间件,它基于Java Message Service (JMS) 规范,提供高效、可靠的消息传递服务。在本文中,我们将深入探讨Apache ActiveMQ,特别是针对“apache-activemq-...

    ActiveMQ路由配置方式

    ActiveMQ路由配置方式 ActiveMQ路由配置是Apache ActiveMQ项目中的一种重要配置方式,它依赖另一个Apache项目Camel。ActiveMQ集成了Camel,启动时同时会启动Camel。通过Camel Web Console可以进行Routing配置。 ...

    apache-activemq-5.9.0 下载

    Apache ActiveMQ是开源社区中最流行的消息中间件之一,它基于Java消息服务(JMS)标准,提供高效、可靠的异步通信解决方案。ActiveMQ在企业级应用中广泛应用,因为它支持多种协议,如OpenWire、STOMP、AMQP、MQTT、...

    ActiveMQ接受和发送工具.rar

    ActiveMQ是中国最流行的开源消息中间件之一,基于Java Message Service(JMS)规范,它提供了一个高可伸缩、高性能、稳定且灵活的消息传递平台。这个"ActiveMQ接受和发送工具.rar"压缩包包含了用于与ActiveMQ交互的...

    7道消息队列ActiveMQ面试题!

    ActiveMQ是一款非常流行的开源消息队列中间件,它实现了JMS(Java Message Service,Java消息服务)1.1规范,面向消息的中间件(Message Oriented Middleware,MOM)是指利用高效可靠的消息传递机制进行与平台无关的...

    ActiveMQ消息服务器 v6.0.1.zip

    ActiveMQ是Apache软件基金会开发的一款开源消息中间件,它遵循开放消息传递标准(JMS,Java Message Service),用于在分布式系统中实现可靠的消息传递。在本文中,我们将深入探讨ActiveMQ v6.0.1的核心特性、应用...

    Spring集成ActiveMQ配置

    Spring集成ActiveMQ是将Spring框架与ActiveMQ消息中间件相结合,实现异步处理和解耦应用程序的关键技术。在本文中,我们将深入探讨如何配置和使用这一组合,以及它在实际项目中的应用。 首先,让我们了解Spring框架...

    activemq自启动并设置用户名密码

    #### 一、ActiveMQ简介 ActiveMQ是一款非常流行的开源消息中间件,它基于Java语言开发,并且遵循了Java消息服务(JMS)规范。ActiveMQ提供了丰富的特性,包括持久化消息存储、事务支持、集群等功能。在企业级应用中,...

    ActiveMQ in Action pdf英文版+源代码

    ActiveMQ in Action pdf英文原版加源代码压缩包。 Apache ActiveMQ in Action is a thorough, practical guide to implementing message-oriented systems in Java using ActiveMQ. The book lays out the core of ...

    apache-activemq Linux版本

    Apache ActiveMQ是业界广泛使用的开源消息中间件,尤其在Linux环境下表现出色。它基于Java语言开发,遵循Apache软件基金会的许可证,并且实现了多种消息传递协议,包括OpenWire、STOMP、AMQP和XMPP等。在Linux系统上...

    最新稳定版ActiveMQ5.15.0

    Apache ActiveMQ是开源社区中最流行的Java消息代理,也是企业级消息中间件(Message Broker)的首选之一。在最新的稳定版5.15.0中,它提供了可靠的消息传递功能,适用于分布式应用程序之间的通信,实现了异步处理、...

    自己实现的ActiveMQ连接池和新版本ActiveMQ自带的连接池,封装好的工具类,可直接使用

    本资源提供的内容是关于ActiveMQ的连接池实现,分为两部分:一是作者自己实现的ActiveMQ连接池,二是新版本ActiveMQ自带的连接池。连接池是一种资源管理技术,通过复用已建立的数据库连接或网络连接,减少创建和销毁...

    springboot集成activemq实现消息接收demo

    而ActiveMQ是Apache出品的一款开源消息中间件,它遵循JMS(Java Message Service)规范,用于处理应用程序之间的异步通信。本教程将详细介绍如何在Spring Boot项目中集成ActiveMQ,实现消息接收的Demo。 首先,我们...

Global site tag (gtag.js) - Google Analytics