您的位置：首页 > 编程语言 > Java开发

Spring与ActiveMQ整合

2017-07-24 21:07 176 查看

Spring与ActiveMQ整合

1. activeMQ准备

既然是使用的apache的activeMQ作为JMS的实现，那么首先我们应该到apache官网上下载activeMQ（http://activemq.apache.org/download.html），进行解压后运行其bin目录下面的activemq.bat文件启动activeMQ。（和tomcat差不多）消息的中转站。

2.配置ConnectionFactory（用于创建链接Connection）

ConnectionFactory是用于产生到JMS服务器的链接的，Spring为我们提供了多个ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory。

SingleConnectionFactory（单例）对于建立JMS服务器链接的请求会一直返回同一个链接，并且会忽略Connection的close方法调用。

CachingConnectionFactory继承了SingleConnectionFactory，所以它拥有SingleConnectionFactory的所有功能，同时它还新增了缓存功能，它可以缓存Session、MessageProducer和MessageConsumer。这里我们使用SingleConnectionFactory为例

<bean id="connectionFactory"

class="org.springframework.jms.connection.SingleConnectionFactory"/>

       这样就定义好产生JMS服务器链接的ConnectionFactory了吗？答案是非也。Spring提供的ConnectionFactory只是Spring用于管理ConnectionFactory的，真正产生到JMS服务器链接的ConnectionFactory还得是由JMS服务厂商提供，并且需要把它注入到Spring提供的ConnectionFactory中。我们这里使用的是ActiveMQ实现的JMS，所以在我们这里真正的可以产生Connection的就应该是由ActiveMQ提供的ConnectionFactory。所以定义一个ConnectionFactory的完整代码应该如下所示：



<bean id="activeMqFactory "

class="org.apache.activemq.ActiveMQConnectionFactory">

<property name="brokerURL" value="tcp://localhost:61616"/>

</bean>



<beanid="cachingMqFactory"
class="org.springframework.jms.connection.CachingConnectionFactory">



<property name="targetConnectionFactory"
ref=" activeMqFactory
"
/>

        <property
name="reconnectOnException"
value="true"
/>

        <property
name="cacheConsumers"
value="false"
/>

        <property name="cacheProducers"
value="false"
/>

        <property
name="sessionCacheSize"
value="50"
/>

</bean>

3.配置连接池

注入connectionFactory属性，对应厂商提供的链接工厂

<bean
id="activeMqPooledFactory"class="org.apache.activemq.pool.PooledConnectionFactory"destroy-
method="stop">

      <property
name="connectionFactory">

         <ref
local=" activeMqFactory
" />

      </property>

</bean>

4. 配置生产者

配置好ConnectionFactory之后我们就需要配置生产者。生产者负责产生消息并发送到JMS服务器，这通常对应的是我们的一个业务逻辑服务实现类。但是我们的
4000
服务实现类是怎么进行消息的发送的呢？这通常是利用Spring为我们提供的JmsTemplate类来实现的，所以配置生产者其实最核心的就是配置进行消息发送的JmsTemplate。对于消息发送者而言，它在发送消息的时候要知道自己该往哪里发，为此，我们在定义JmsTemplate的时候需要往里面注入一个Spring提供的ConnectionFactory对象

此处若使用链接池则注入链接池bean（activeMqPooledFactory），不使用链接池注入

cachingMqFactory（CachingConnectionFactory或SingleConnectionFactory）



<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

    

    <property name="connectionFactory" ref="activeMqPooledFactory"/>

</bean>

5.配置消息发送目的地（Destination）

JmsTemplate进行消息发送的时候，我们需要知道消息发送的目的地，即destination。在Jms中有一个用来表示目的地的Destination接口，它里面没有任何方法定义，只是用来做一个标识而已。当我们在使用JmsTemplate进行消息发送时没有指定destination的时候将使用默认的Destination。默认Destination可以通过在定义jmsTemplate bean对象时通过属性defaultDestination或defaultDestinationName来进行注入，defaultDestinationName对应的就是一个普通字符串。在ActiveMQ中实现了两种类型的Destination，一个是点对点的ActiveMQQueue，另一个就是支持订阅/发布模式的ActiveMQTopic。在定义这两种类型的Destination时我们都可以通过一个name属性来进行构造：



<bean id="queueDestination"

class="org.apache.activemq.command.ActiveMQQueue">

       <constructor-arg>

                <value>queue</value>  队列名称

        </constructor-arg>

</bean>



<bean id="topicDestination"

class="org.apache.activemq.command.ActiveMQTopic">

<constructor-arg value="topic"/> 主题名称

</bean>

6.配置消费者

生产者往指定目的地Destination发送消息后，接下来就是消费者对指定目的地的消息进行消费了。那么消费者是如何知道有生产者发送消息到指定目的地Destination了呢？这是通过Spring为我们封装的消息监听容器MessageListenerContainer实现的，它负责接收信息，并把接收到的信息分发给真正的MessageListener进行处理。每个消费者对应每个目的地都需要有对应的MessageListenerContainer。对于消息监听容器而言，除了要知道监听哪个目的地之外，还需要知道到哪里去监听，也就是说它还需要知道去监听哪个JMS服务器，这是通过在配置MessageConnectionFactory的时候往里面注入一个ConnectionFactory来实现的。所以我们

在配置一个MessageListenerContainer的时候有三个属性必须指定，

一个是表示从哪里监听的ConnectionFactory；

一个是表示监听什么的Destination；

一个是接收到消息以后进行消息处理的MessageListener。

Spring提供了两种类型的MessageListenerContainer，SimpleMessageListenerContainer

DefaultMessageListenerContainer。

SimpleMessageListenerContainer会在一开始的时候就创建一个会话session和消费者Consumer，并且会使用标准的JMSMessageConsumer.setMessageListener()方法注册监听器让JMS提供者调用监听器的回调函数。它不会动态的适应运行时需要和参与外部的事务管理。兼容性方面，它非常接近于独立的JMS规范，但一般不兼容Java
EE的JMS限制。

大多数情况还是使用的DefaultMessageListenerContainer，跟SimpleMessageListenerContainer相比，DefaultMessageListenerContainer会动态的适应运行时需要，并且能够参与外部的事务管理。它很好的平衡了对JMS提供者要求低、先进功能如事务参与和兼容Java
EE环境。

定义处理消息的MessageListener

       要定义处理消息的MessageListener我们只需要实现JMS规范中的MessageListener接口就可以了。MessageListener接口中只有一个方法onMessage方法，当接收到消息的时候会自动调用该方法。



<bean id="queueDestination"

class="org.apache.activemq.command.ActiveMQQueue">

    <constructor-arg>

        <value>queue</value>

    </constructor-arg>

</bean>



<bean id="consumerMessageListener"

class="com.tiantian.springintejms.listener.ConsumerMessageListener"/>



<bean id="jmsContainer"

class="org.springframework.jms.listener.DefaultMessageListenerContainer">

    <property name="connectionFactory" ref="connectionFactory" />

    <property name="destination" ref="queueDestination" />

    <property name="messageListener" ref="consumerMessageListener" />

</bean>

我们可以看到我们定义了一个名叫queue的ActiveMQQueue目的地，我们的监听器就是监听了发送到这个目的地的消息。至此我们的生成者和消费者都配置完成了，这也就意味着我们的整合已经完成了。这个时候完整的Spring的配置文件应该是这样的：

<?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:context="http://www.springframework.org/schema/context"

    xmlns:jms="http://www.springframework.org/schema/jms"

    xsi:schemaLocation="http://www.springframework.org/schema/beans

     http://www.springframework.org/schema/beans/spring-beans-3.0.xsd

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

     http://www.springframework.org/schema/context/spring-context-3.0.xsd

    http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd

    http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd">

    <context:component-scan base-package="com.tiantian" />

    

<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

        

        <property name="connectionFactory" ref="connectionFactory"/>

    </bean>  [/b]



    

    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

        <property name="brokerURL" value="tcp://localhost:61616"/>

    </bean>



    

    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">

        

    <property name="targetConnectionFactory" ref="targetConnectionFactory"/>

    </bean>



    

    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">

        <constructor-arg>

            <value>queue</value>

        </constructor-arg>

    </bean>

    

<bean id="consumerMessageListener"

class="com.tiantian.springintejms.listener.ConsumerMessageListener"/>

    

  <bean id="jmsContainer"

   class="org.springframework.jms.listener.DefaultMessageListenerContainer">

        <property name="connectionFactory" ref="connectionFactory" />

        <property name="destination" ref="queueDestination" />

        <property name="messageListener" ref="consumerMessageListener" />

    </bean>

</beans>

7.消息监听器MessageListener

在spring整合JMS的应用中我们在定义消息监听器的时候一共可以定义三种类型的消息监听器，分别是MessageListener、SessionAwareMessageListener和MessageListenerAdapter。下面就分别来介绍一下这几种类型的区别。

7.1 MessageListener

MessageListener是最原始的消息监听器，它是JMS规范中定义的一个接口。其中定义了一个用于处理接收到的消息的onMessage方法，该方法只接收一个Message参数。我们前面在讲配置消费者的时候用的消息监听器就是MessageListener，代码如下：

1.     import javax.jms.JMSException;

2.     import javax.jms.Message;

3.     import javax.jms.MessageListener;

4.     import javax.jms.TextMessage;

5.     public class ConsumerMessageListener implements MessageListener {

6.

7.         public void onMessage(Message message) {

8.             //这里我们知道生产者发送的就是一个纯文本消息，所以这里可以直接进行强制转换，或者直接把onMessage方法的参数改成Message的子类TextMessage

9.             TextMessage textMsg = (TextMessage) message;

10.         System.out.println("接收到一个纯文本消息。");

11.         try {

12.             System.out.println("消息内容是：" + textMsg.getText());

13.         } catch (JMSException e) {

14.             e.printStackTrace();

15.         }

16.     }

17. }

7.2 SessionAwareMessageListener

SessionAwareMessageListener是Spring为我们提供的，它不是标准的JMSMessageListener。MessageListener的设计只是纯粹用来接收消息的，假如我们在使用MessageListener处理接收到的消息时我们需要发送一个消息通知对方我们已经收到这个消息了，那么这个时候我们就需要在代码里面去重新获取一个Connection或Session。SessionAwareMessageListener的设计就是为了方便我们在接收到消息后发送一个回复的消息，它同样为我们提供了一个处理接收到的消息的onMessage方法，但是这个方法可以同时接收两个参数，一个是表示当前接收到的消息Message，另一个就是可以用来发送消息的Session对象。先来看一段代码：

1.     package com.tiantian.springintejms.listener;

2.     import javax.jms.Destination;

3.     import javax.jms.JMSException;

4.     import javax.jms.Message;

5.     import javax.jms.MessageProducer;

6.     import javax.jms.Session;

7.     import javax.jms.TextMessage;

8.     import org.springframework.jms.listener.SessionAwareMessageListener;

9.     public class ConsumerSessionAwareMessageListener implements

10.          SessionAwareMessageListener<TextMessage> {

11.     private Destination destination;

12.

13.

14. public void onMessage(TextMessage message, Session session) throws JMSException {

15.         System.out.println("收到一条消息");

16.         System.out.println("消息内容是：" + message.getText());

17.         MessageProducer producer = session.createProducer(destination);

18.         Message textMessage = session.createTextMessage("ConsumerSessionAwareMessageListener");

19.         producer.send(textMessage);

20.     }

21.     public Destination getDestination() {

22.         returndestination;

23.     }

24.     public void setDestination(Destination destination) {

25.         this.destination = destination;

26.     }

27. }

在上面代码中我们定义了一个SessionAwareMessageListener，在这个Listener中我们在接收到了一个消息之后，利用对应的Session创建了一个到destination的生产者和对应的消息，然后利用创建好的生产者发送对应的消息。

接着我们在Spring的配置文件中配置该消息监听器将处理来自一个叫sessionAwareQueue的目的地的消息，并且往该MessageListener中通过set方法注入其属性destination的值为queueDestination。这样当我们的SessionAwareMessageListener接收到消息之后就会往queueDestination发送一个消息。

1.     <?xml version="1.0" encoding="UTF-8"?>

2.     <beans xmlns="http://www.springframework.org/schema/beans"

3.         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"

4.         xmlns:jms="http://www.springframework.org/schema/jms"

5.         xsi:schemaLocation="http://www.springframework.org/schema/beans

6.          http://www.springframework.org/schema/beans/spring-beans-3.0.xsd

7.          http://www.springframework.org/schema/context

8.          http://www.springframework.org/schema/context/spring-context-3.0.xsd

9.         http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd

10.       http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd">

11.

12.       <context:component-scan base-package="com.tiantian" />

13.       

14.       <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

15.           

16.           <property name="connectionFactory" ref="connectionFactory"/>

17.       </bean>

18.

19.       

20.       <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

21.           <property name="brokerURL" value="tcp://localhost:61616"/>

22.       </bean>

23.

24.       

25.       <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">

26.           

27.           <property name="targetConnectionFactory" ref="targetConnectionFactory"/>

28.       </bean>

29.

30.       

31.       <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">

32.           <constructor-arg>

33.               <value>queue</value>

34.           </constructor-arg>

35.       </bean>

36.       

37.       <bean id="sessionAwareQueue" class="org.apache.activemq.command.ActiveMQQueue">

38.           <constructor-arg>

39.               <value>sessionAwareQueue</value>

40.           </constructor-arg>

41.       </bean>

42.       

43.       <bean id="consumerMessageListener" class="com.tiantian.springintejms.listener.ConsumerMessageListener"/>

44.       

45.       <bean id="consumerSessionAwareMessageListener" class="com.tiantian.springintejms.listener.ConsumerSessionAwareMessageListener">

46.           <property name="destination" ref="queueDestination"/>

47.       </bean>

48.       

49.       <bean id="jmsContainer"        class="org.springframework.jms.listener.DefaultMessageListenerContainer">

50.           <property name="connectionFactory" ref="connectionFactory" />

51.           <property name="destination" ref="queueDestination" />

52.           <property name="messageListener" ref="consumerMessageListener" />

53.       </bean>

54.

55.       <bean id="sessionAwareListenerContainer"

56.           class="org.springframework.jms.listener.DefaultMessageListenerContainer">

57.           <property name="connectionFactory" ref="connectionFactory" />

58.           <property name="destination" ref="sessionAwareQueue" />

59.           <property name="messageListener" ref="consumerSessionAwareMessageListener" />

60.       </bean>

61.   </beans>

这说明我们已经成功的往sessionAwareQueue发送了一条纯文本消息，消息会被ConsumerSessionAwareMessageListener的onMessage方法进行处理，在onMessage方法中ConsumerSessionAwareMessageListener就是简单的把接收到的纯文本信息的内容打印出来了，之后再往queueDestination发送了一个纯文本消息，消息内容是“ConsumerSessionAwareMessageListener…”，该消息随后就被ConsumerMessageListener处理了，根据我们的定义，在ConsumerMessageListener中也只是简单的打印了一下接收到的消息内容。

7.3 MessageListenerAdapter

MessageListenerAdapter类实现了MessageListener接口和SessionAwareMessageListener接口，它的主要作用是将接收到的消息进行类型转换，然后通过反射的形式把它交给一个普通的Java类进行处理。

      MessageListenerAdapter会把接收到的消息做如下转换：

TextMessage转换为String对象；

BytesMessage转换为byte数组；

MapMessage转换为Map对象；

      ObjectMessage转换为对应的Serializable对象。

      既然前面说了MessageListenerAdapter会把接收到的消息做一个类型转换，然后利用反射把它交给真正的目标处理器——一个普通的Java类进行处理（如果真正的目标处理器是一个MessageListener或者是一个SessionAwareMessageListener，那么Spring将直接使用接收到的Message对象作为参数调用它们的onMessage方法，而不会再利用反射去进行调用），那么我们在定义一个MessageListenerAdapter的时候就需要为它指定这样一个目标类。这个目标类我们可以通过MessageListenerAdapter的构造方法参数指定，如：



<bean id="messageListenerAdapter"

class="org.springframework.jms.listener.adapter.MessageListenerAdapter">

        <constructor-arg>

            <bean class="com.tiantian.springintejms.listener.ConsumerListener"/>

        </constructor-arg>

    </bean>

也可以通过它的delegate属性来指定，如：



<bean id="messageListenerAdapter"

class="org.springframework.jms.listener.adapter.MessageListenerAdapter">

        <property name="delegate">

            <bean class="com.tiantian.springintejms.listener.ConsumerListener"/>

        </property>

        <property name="defaultListenerMethod" value="receiveMessage"/>

</bean>

如果我们指定的这个目标处理器是一个MessageListener或者是一个SessionAwareMessageListener的时候Spring将直接利用接收到的Message对象作为方法参数调用它们的onMessage方法。但是如果指定的目标处理器是一个普通的Java类时Spring将利用Message进行了类型转换之后的对象作为参数通过反射去调用真正的目标处理器的处理方法，那么Spring是如何知道该调用哪个方法呢？这是通过MessageListenerAdapter的defaultListenerMethod属性来决定的，当我们没有指定该属性时，Spring会默认调用目标处理器的handleMessage方法。

      示例，假设我们有一个普通的Java类ConsumerListener，其对应有两个方法，handleMessage和receiveMessage，其代码如下：

package com.tiantian.springintejms.listener;

public class ConsumerListener {

    public void handleMessage(String message) {

        System.out.println("ConsumerListener通过handleMessage接收到一个纯文本消息，消息内容是：" + message);

    }

    public void receiveMessage(String message) {

        System.out.println("ConsumerListener通过receiveMessage接收到一个纯文本消息，消息内容是：" + message);

    }

}

消息创建MyMessageCreator

package jms.activemq.myexample.spring;

import java.util.Date;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.Session;

import javax.jms.TextMessage;

import org.springframework.jms.core.MessageCreator;

public class MyMessageCreator
implements MessageCreator {

    /**

     * 消息序号

     */

    private int msgNo;

    public MyMessageCreator(int no)
{

        this.msgNo = no;

    }

    @Override

    public Message createMessage(Session session) throws JMSException
{

      TextMessage textMsg = session.createTextMessage();

       textMsg.setText(new Date() + "第" +
this.msgNo + "条消息发出");

      return textMsg;

    }

}

异步消息监听器TextListener

package jms.activemq.myexample.spring;

import javax.jms.*;

/**

* Text消息监听

*

*/

public class TextListener
implements MessageListener {

    /**

     * Casts the message to a TextMessage and displays its text.

     * @param message

     * the incoming message

     */

    public void onMessage(Message
message) {

        TextMessage msg = null;

        try {

            if (message instanceof TextMessage)
{

                msg = (TextMessage) message;

        System.out.println("Reading message: " + msg.getText());

            } else {

        System.out.println("Message of wrong type: "

                    + message.getClass().getName());

            }

        } catch (JMSException e) {

            System.out.println("JMSException in onMessage(): " + e.toString());

        } catch (Throwable t) {

            System.out.println("Exception in onMessage():" + t.getMessage());

        }

    }

}

测试程序SpringJmsTestMain

package jms.activemq.myexample.spring;

import org.springframework.context.ApplicationContext;

import org.springframework.context.support.ClassPathXmlApplicationContext;

public class SpringJmsTestMain
{

    public static void main(String[]
args) throws InterruptedException {

        ApplicationContext context = new ClassPathXmlApplicationContext(

                new String[] { "SpringJms/SpringJms.xml" });

        SpringPublisher publisher = (SpringPublisher) context

                .getBean("springPublisher");

        publisher.start();

    }

}

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"

    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
  <!—初始化jms连接工厂（厂商提供）-->

    <beanid="jmsFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

        <property name="brokerURL">

            <value>tcp://localhost:61616</value>

        </property>

    </bean>







    

    

    <bean id="myTopic" class="org.apache.activemq.command.ActiveMQTopic"

        autowire="constructor">

        <constructor-arg value="STOCKS.JAVA" />

    </bean>

    

    <bean id="myTextListener" class="jms.activemq.myexample.spring.TextListener">

    </bean>

    

    <bean id="javaConsumer"

        class="org.springframework.jms.listener.DefaultMessageListenerContainer">

        <property name="connectionFactory" ref="jmsFactory" />

        <property name="destination" ref="myTopic" />

        <property name="messageListener" ref="myTextListener" />

    </bean>

    

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

        <property name="connectionFactory">

            <ref local="jmsFactory" />

        </property>

    </bean>

    

    <bean id="springPublisher" class="jms.activemq.myexample.spring.SpringPublisher">

        <property name="template">

            <ref local="jmsTemplate" />

        </property>

        <property name="topic">

            <ref local="myTopic" />

        </property>

    </bean>

</beans>

内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理

标签：

相关文章推荐

新的分享

章节导航