淺析JMS 之 Active MQ 的spring整合

一、與spring整合實作ptp的同步接收訊息

pom.xml:

   <!--  --><dependency><groupId>org.springframework</groupId><artifactId>spring-jms</artifactId><version>4.3.7.RELEASE</version></dependency>
  <!--  --><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-pool</artifactId><version>5.9.0</version></dependency>

spring-jms.xml:

<?xml version="1.0" encoding="UTF-8"?><beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:jms="http://www.springframework.org/schema/jms"xmlns:amq="http://activemq.apache.org/schema/core"xsi:schemaLocation="http://activemq.apache.org/schema/core

http://www.springframework.org/schema/jms

http://www.springframework.org/schema/beans
"><!-- ActiveMQConnectionFactory就是JMS中负责创建到ActiveMQ连接的工厂类 --><bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory" > <property name="brokerURL" value="tcp://192.168.0.224:61616"/>  </bean><!-- 创建连接池 --><bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">  <property name="connectionFactory" ref="connectionFactory"/>  <property name="maxConnections" value="10"/>  </bean>  <!-- Spring为我们提供了多个ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory --><bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">  <property name="targetConnectionFactory" ref="pooledConnectionFactory"/>  </bean> <!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->  <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  <property name="connectionFactory" ref="cachingConnectionFactory"/>  </bean> <!--这个是队列目的地，点对点的-->  <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">  <constructor-arg index="0" value="spring-queue"/> </bean>   </beans>

   ConnectionFactory是用於產生到JMS伺服器的連結的，Spring為我們提供了多個ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory。 SingleConnectionFactory對於建立JMS伺服器連結的請求會一直傳回同一個鏈接，並且會忽略Connection的close方法呼叫。 CachingConnectionFactory繼承了SingleConnectionFactory，所以它擁有SingleConnectionFactory的所有功能，同時它也新增了快取功能，它可以快取Session、MessageProducer和MessageConsumer。這裡我們使用CachingConnectionFactory來作為範例。

訊息生產者：

   = ClassPathXmlApplicationContext("spring-jms.xml"=(JmsTemplate) context.getBean("jmsTemplate"=(Destination) context.getBean("queueDestination" Message createMessage(Session session)  session.createTextMessage("Hello spring JMS"

#消費者：

package com.jalja.org.jms.spring;import javax.jms.Destination;import org.springframework.context.ApplicationContext;import org.springframework.context.support.ClassPathXmlApplicationContext;import org.springframework.jms.core.JmsTemplate;public class SpringJmsReceive {public static void main(String[] args) {
    ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
    JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
    Destination queueDestination=(Destination) context.getBean("queueDestination");
    String msg=(String) jmsTemplate.receiveAndConvert(queueDestination);
    System.out.println(msg);
　　}
}

二、PTP的非同步呼叫

 我們在spring中直接配置非同步接收訊息的監聽器，這樣就相當於在spring中配置了消費者，在接受訊息的時候就不必要啟動消費者了。

spring-jms.xml:

<?xml version="1.0" encoding="UTF-8"?><beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:jms="http://www.springframework.org/schema/jms"xmlns:amq="http://activemq.apache.org/schema/core"xsi:schemaLocation="http://activemq.apache.org/schema/core

http://www.springframework.org/schema/jms

http://www.springframework.org/schema/beans
"><!-- ActiveMQConnectionFactory就是JMS中负责创建到ActiveMQ连接的工厂类 --><bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory" > <property name="brokerURL" value="tcp://192.168.0.224:61616"/>  </bean><!-- 创建连接池 --><bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">  <property name="connectionFactory" ref="connectionFactory"/>  <property name="maxConnections" value="10"/>  </bean>  <!-- Spring为我们提供了多个ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory --><bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">  <property name="targetConnectionFactory" ref="pooledConnectionFactory"/>  </bean> <!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->  <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  <property name="connectionFactory" ref="cachingConnectionFactory"/>  </bean> <!--这个是队列目的地，点对点的-->  <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">  <constructor-arg index="0" value="spring-queue"/> </bean>   <!-- 消息监听器 -->  <bean id="myMessageListener" class="com.jalja.org.jms.spring.yb.MyMessageListener"/>  <!-- 消息监听容器 -->  <bean id="jmsContainer"  class="org.springframework.jms.listener.DefaultMessageListenerContainer">  <property name="connectionFactory" ref="cachingConnectionFactory" />  <property name="destination" ref="queueDestination" />  <property name="messageListener" ref="myMessageListener" />  </bean> </beans>

　　生產者往指定目的地Destination發送訊息後，接下來就是消費者對指定目的地的訊息進行消費了。那麼消費者是如何知道有生產者傳送訊息到指定目的地Destination了呢？這是透過Spring為我們封裝的訊息監聽容器MessageListenerContainer實現的，它負責接收訊息，並把接收到的訊息分發給真正的MessageListener進行處理。每個消費者對應每個目的地都需要有對應的MessageListenerContainer。對於訊息監聽容器而言，除了要知道監聽哪個目的地之外，還需要知道到哪裡去監聽，也就是說它還需要知道去監聽哪個JMS伺服器，這是透過在配置MessageConnectionFactory的時候往裡面注入一個ConnectionFactory來實現的。所以我們在設定一個MessageListenerContainer的時候有三個屬性必須指定，一個是表示從哪裡監聽的ConnectionFactory；一個是表示監聽什麼的Destination；一個是接收到訊息以後進行訊息處理的MessageListener。 Spring總共為我們提供了兩種類型的MessageListenerContainer，SimpleMessageListenerContainer和DefaultMessageListenerContainer。
　　SimpleMessageListenerContainer：SimpleMessageListenerContainer會在一開始的時候就建立一個會話session和消費者Consumer，並且會使用標準的JMS MessageConsumer.setMessageListener()方法註冊監聽器讓JMS提供者呼叫監聽器的回調函數。它不會動態的適應執行時間需要和參與外部的事務管理。在相容性方面，它非常接近獨立的JMS規範，但一般不相容Java EE的JMS限制。

　　DefaultMessageListenerContainer：在大多數情況下我們還是使用的DefaultMessageListenerContainer，跟SimpleMessageListenerContainer相比，DefaultMessageListenerContainer會動態的適應運行時需要，並且能夠參與外部的事務管理。它很好的平衡了對JMS提供者要求低、先進功能如事務參與和兼容Java EE環境。

訊息生產者：

    public static void main(String[] args) {
        ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
        JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
        Destination queueDestination=(Destination) context.getBean("queueDestination");
        System.out.println("异步调用执行开始");
        jmsTemplate.send(queueDestination, new MessageCreator(){
            @Overridepublic Message createMessage(Session session) throws JMSException {return session.createTextMessage("Hello spring JMS");
            }
        });
        System.out.println("异步调用执行结束");
    }

訊息監聽器：MyMessageListener

public class MyMessageListener implements MessageListener{
    @Overridepublic void onMessage(Message message) {
        TextMessage msg= (TextMessage) message;try {
            System.out.println("你好："+msg.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

啟動訊息生產者 監聽器的執行結果是：

异步调用执行开始
异步调用执行结束
你好：Hello spring JMS

三、發布訂閱同步接收

#spring-jms.xml:

<?xml version="1.0" encoding="UTF-8"?><beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:jms="http://www.springframework.org/schema/jms"xmlns:amq="http://activemq.apache.org/schema/core"xsi:schemaLocation="http://activemq.apache.org/schema/core

http://www.springframework.org/schema/jms

http://www.springframework.org/schema/beans
"><!-- ActiveMQConnectionFactory就是JMS中负责创建到ActiveMQ连接的工厂类 --><bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory" > <property name="brokerURL" value="tcp://192.168.0.224:61616"/>  </bean><!-- 创建连接池 --><bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">  <property name="connectionFactory" ref="connectionFactory"/>  <property name="maxConnections" value="10"/>  </bean>  <!-- Spring为我们提供了多个ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory --><bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">  <property name="targetConnectionFactory" ref="pooledConnectionFactory"/>  </bean> <!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->  <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  <property name="connectionFactory" ref="cachingConnectionFactory"/>  </bean> <!--这个是队列目的地，发布订阅-->  <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">  <constructor-arg index="0" value="spring-Topic"/> </bean>   </beans>

#生產者：

    public static void main(String[] args) {
        ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
        JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
        Destination topicDestination=(Destination) context.getBean("topicDestination");
        jmsTemplate.send(topicDestination, new MessageCreator(){
            @Overridepublic Message createMessage(Session session) throws JMSException {return session.createTextMessage("Hello spring JMS topicDestination");
            }
        });
    }

消費者：

public class SpringJmsSubscriber {public static void main(String[] args) {
        ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
        JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
        Destination topicDestination=(Destination) context.getBean("topicDestination");
        String msg=(String) jmsTemplate.receiveAndConvert(topicDestination);
        System.out.println(msg);
    }
}

 

#

以上是淺析JMS 之 Active MQ 的spring整合的詳細內容。更多資訊請關注PHP中文網其他相關文章！