ActiveMQ

guomingzhang2008

浏览: 160780 次
性别:
来自: 上海

最近访客更多访客>>

旺仔小呆

lifan51777

qq849397558

zo1698

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

activemq

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

分享到：