Java中间件JMS(二)之ActiveMQ整合spring(一)

原文链接：http://blog.csdn.net/dwc_fly/article/details/10306805 

 

在上一章( Java中间件JMS之ActiveMQ入门http://blog.csdn.net/dengwanchuan/article/details/10241345)说到ActiveMQ能与spring进行整合,ActiveMQ与Spring进行整合有一定的好处,首先是可配置化,然后是能使用Spring的aop,tx等特性进行项目开发.

一.准备工作

我使用的是spring版本是4.0.0.M2，其他版本的也可以，只是配置不同，去Spring官网下载zip包，解开后将dist目录下的所有jar包（根据自己选择）拷贝到项目lib目录下并加入项目项目中的lib中，一般jms所需要的Spring的jar有:

二.代码开发

1.在src目录下新建applicationContext.xml文件并输入一下内容：

<?xml version="1.0" encoding="GBK"?>
<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.xsd">
</beans>

 

2.引入spring,打开web.xml并将其内容修改为以下内容：

<?xml version="1.0" encoding="GBK"?>
<web-app version="2.5" xmlns="http://java.sun.com/xml/ns/javaee"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://java.sun.com/xml/ns/javaee
    http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd">
    <context-param>
        <param-name>contextConfigLocation</param-name>
        <param-value>classpath*:applicationContext*.xml</param-value>
    </context-param>
    <servlet>
        <servlet-name>spring</servlet-name>
        <servlet-class>
            org.springframework.web.servlet.DispatcherServlet
        </servlet-class>
        <load-on-startup>1</load-on-startup>
    </servlet>
    <servlet-mapping>
        <servlet-name>spring</servlet-name>
        <url-pattern>/</url-pattern>
    </servlet-mapping>
    <welcome-file-list>
        <welcome-file>index.jsp</welcome-file>
    </welcome-file-list>
</web-app>

3.配置JMSTemplate模板

 

类似于jdbcTemplate，首先要配置一个ConnectionFactory，之后要开始配置JmsTemplate模板了。最后是配置消息目标了。消息分为队列(Queue)和主题(Topic)两大类。在applicationContext.xml中加入如下内容：

 

<!-- 配置JMS连接工厂 -->
    <bean id="connectionFactory"
        class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://localhost:61616" />
    </bean>
    <!-- 发送消息的目的地（队列） -->
    <bean id="queueDest"
        class="org.apache.activemq.command.ActiveMQQueue">
        <!-- 设置消息队列的名字 -->
        <constructor-arg index="0" value="myQueue" />
    </bean>
    <!-- 配置Jms模板  -->
    <bean id="jmsQueueTemplate"
        class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="defaultDestination" ref="queueDest" />
        <!--<property name="receiveTimeout" value="10000" />  -->
    </bean>

    <!-- 发送消息的目的地（主题） -->
    <bean id="topicDest"
        class="org.apache.activemq.command.ActiveMQTopic">
        <!-- 设置消息队列的名字 -->
        <constructor-arg index="0" value="myTopic" />
    </bean>
    <!-- 配置TopicJms模板  -->
    <bean id="jmsTopicTemplate"
        class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="defaultDestination" ref="topicDest" />
        <!-- 配置是否为发布订阅者模式，默认为false -->
        <property name="pubSubDomain" value="true"/>
    <!--<property name="receiveTimeout" value="10000" />  -->
    </bean>

 

receiveTimeout表示接收消息时的超时时间，设置的为10秒，因为如果不设置的话，加入接收消息时是阻塞着的，那么将一直阻塞下去。配置完成了。但是我不建议设置这个时间,如果到达设置时间之后,生产者没有运行,消费者接受到Message对象为null,测试可能会出现异常，而且消费者将停止接受消息.那么如何使用JmsTemplate发送消息呢?

spring的beanfactory得到一个jmsTemplate的实例和消息目标的实例，发送消息，够简单的吧。首先我们还从queue方式开始。下面我们就来编写具体代码。

4、编写Point-to-Point (点对点)代码

新建生产者类QueueProducerService.java，代码如下：

 

package jms.mq.spring;

import java.util.Date;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

public class QueueProducerService{
    JmsTemplate jmsTemplate;

    Destination destination;

    public void send() {
        MessageCreator messageCreator = new MessageCreator() {
            public Message createMessage(Session session) throws JMSException {
                TextMessage message = session.createTextMessage();
                message.setText("QueueProducerService发送消息"+new Date());
                return message;
            }

        };
        jmsTemplate.send(this.destination,messageCreator);
    }

    public void setJmsTemplate(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }

    public void setDestination(Destination destination) {
        this.destination = destination;
    }
}

 

生产者编写完了，下面我们来编写消费者，上面说了，发送消息的时候，spring的beanfactory得到一个jmsTemplate的实例和消息目标的实例，然后发送，那么接受的时候肯定也是得到一个jmsTemplate的实例和消息目标的实例，然后接受，下面我们来看具体代码。

新建一个消费者类QueueConsumerService.java，具体代码如下：

 

package jms.mq.spring;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.TextMessage;

import org.springframework.jms.core.JmsTemplate;


public class QueueConsumerService{

    JmsTemplate jmsTemplate;

    Destination destination;

    public void receive() {
        TextMessage message = (TextMessage) jmsTemplate.receive();
        try {
            System.out.println("QueueConsumerService收到消息："+message.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    public void setJmsTemplate(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }

    public void setDestination(Destination destination) {
        this.destination = destination;
    }
}

 

代码编写完毕，下面要进行bean的配置，在applicationContext.xml中加入如下代码实例化对象和依赖注入:

 

<bean id="queueProducerService" class="jms.mq.spring.QueueProducerService">
    <property name="jmsTemplate" ref="jmsQueueTemplate" />
    <property name="destination" ref="queueDest" />
</bean>

<bean id="queueConsumerService" class="jms.mq.spring.QueueConsumerService">
    <property name="jmsTemplate" ref="jmsQueueTemplate" />
    <property name="destination" ref="queueDest" />
</bean>

 

需要的业务代码都已编写完毕，下面编写测试代码。新建一个生产者的测试类QueueProducerTest.java。具体代码如下：

package jms.mq.spring;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class QueueProducerTest {
    private static ApplicationContext appContext = new ClassPathXmlApplicationContext( "applicationContext.xml");

    private static void send() {
        QueueProducerService producerService = (QueueProducerService) appContext.getBean("queueProducerService");
        producerService.send();
    }

    /**
     * @param args
     */
    public static void main(String[] args) {
        send();
    }

}

再建一个消费者的测试类，QueueConsumerTest.java，具体代码如下：

 

package jms.mq.spring;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class QueueConsumerTest {
    private static ApplicationContext appContext = new ClassPathXmlApplicationContext( "applicationContext.xml");

    private static void receive() {
        QueueConsumerService consumerService = (QueueConsumerService) appContext.getBean("queueConsumerService");
        consumerService.receive();
    }

    public static void main(String[] args) {
        receive();
    }

}

 

 

5、运行point-point(点对点)程序

所有代码都编写完了，我们来看一下我们的劳动成果。运行生产者测试类。控制台打印出如下内容，画线标注的就是我们发送的内容：

6、编写Publisher/Subscriber(发布/订阅者)代码

新建发布者TopicPublisherService.java，内容如下：

package jms.mq.spring;

import java.util.Date;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

import jms.spring.QueueProducerService;

public class TopicPublisherService{
    JmsTemplate jmsTemplate;

    Destination destination;

    public void send() {
        MessageCreator messageCreator = new MessageCreator() {
            public Message createMessage(Session session) throws JMSException {
                TextMessage message = session.createTextMessage();
                message.setText("QueueProducerService发送消息"+new Date());
                return message;
            }
        };
        jmsTemplate.send(this.destination,messageCreator);

    }

    public void setJmsTemplate(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }

    public void setDestination(Destination destination) {
        this.destination = destination;
    }

}

再新建一个订阅者TopicSubscriberService.java,代码如下。

package jms.mq.spring;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.TextMessage;

import org.springframework.jms.core.JmsTemplate;

import jms.spring.QueueConsumerService;

public class TopicSubscriberService{

    JmsTemplate jmsTemplate;

    Destination destination;

    public void receive() {
        TextMessage message = (TextMessage) jmsTemplate.receive();
        try {
            System.out.println("QueueConsumerService收到消息："+message.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    public void setJmsTemplate(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }

    public void setDestination(Destination destination) {
        this.destination = destination;
    }
}

在配置文件中applicationContext.xml增加如下配置：

<span style="white-space:pre">  </span><bean id="topicPublisherService" class="jms.mq.spring.TopicPublisherService">
       <property name="jmsTemplate" ref="jmsTopicTemplate"/>
       <property name="destination" ref="topicDest"/>
    </bean>

    <bean id="topicSubscriberService" class="jms.mq.spring.TopicSubscriberService">
       <property name="jmsTemplate" ref="jmsTopicTemplate"/>
       <property name="destination" ref="topicDest"/>
    </bean>

 

编写测试程序发布者测试类，TopicPublisherTest.java

 

package jms.mq.spring;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class TopicPublisherTest {
    private static ApplicationContext appContext = new ClassPathXmlApplicationContext( "applicationContext.xml");

    private static void send() {
        TopicPublisherService topicPublisherService = (TopicPublisherService) appContext.getBean("topicPublisherService");
        topicPublisherService.send();
    }

    public static void main(String[] args) {
        send();
    }
}

编写测试程序订阅者测试类，TopicSubscriberTest.java

package jms.mq.spring;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class TopicSubscriberTest {
    private static ApplicationContext appContext = new ClassPathXmlApplicationContext( "applicationContext.xml");

    private static void receive() {
        TopicSubscriberService topicSubscriberService = (TopicSubscriberService) appContext.getBean("topicSubscriberService");
        topicSubscriberService.receive();
    }

    public static void main(String[] args) {
        receive();
    }
}

7.Publisher/Subscriber(发布/订阅者)程序

先运行订阅者，再运行发布者，可以看到订阅者能打印信息；但是反之就不行，这就是Publisher/Subscriber(发布/订阅者)的特性;

跟Point-Point(点对点)对比的话,不管运行生存者还是消费者，都会打印信息,可以阅读前一章http://blog.csdn.net/dengwanchuan/article/details/10241345了解这两种模式的区别和联系。

 

 

附加完整的applicationContext.xml配置文件

 

<?xml version="1.0" encoding="GBK"?>
<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.xsd">
    <!-- 配置JMS连接工厂 -->
    <bean id="connectionFactory"
        class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://localhost:61616" />
    </bean>
    <!-- 发送消息的目的地（队列） -->
    <bean id="queueDest"
        class="org.apache.activemq.command.ActiveMQQueue">
        <!-- 设置消息队列的名字 -->
        <constructor-arg index="0" value="myQueue" />
    </bean>
    <!-- 配置Jms模板  -->
    <bean id="jmsQueueTemplate"
        class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="defaultDestination" ref="queueDest" />
        <!--<property name="receiveTimeout" value="10000" />  -->
    </bean>

    <!-- 发送消息的目的地（主题） -->
    <bean id="topicDest"
        class="org.apache.activemq.command.ActiveMQTopic">
        <!-- 设置消息队列的名字 -->
        <constructor-arg index="0" value="myTopic" />
    </bean>
    <!-- 配置TopicJms模板  -->
    <bean id="jmsTopicTemplate"
        class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="defaultDestination" ref="topicDest" />
        <!-- 配置是否为发布订阅者模式，默认为false -->
        <property name="pubSubDomain" value="true"/>
    <!--<property name="receiveTimeout" value="10000" />  -->
    </bean>

    <bean id="queueProducerService" class="jms.mq.spring.QueueProducerService">
        <property name="jmsTemplate" ref="jmsQueueTemplate" />
        <property name="destination" ref="queueDest" />
    </bean>

    <bean id="queueConsumerService" class="jms.mq.spring.QueueConsumerService">
        <property name="jmsTemplate" ref="jmsQueueTemplate" />
        <property name="destination" ref="queueDest" />
    </bean>


    <bean id="topicPublisherService" class="jms.mq.spring.TopicPublisherService">
       <property name="jmsTemplate" ref="jmsTopicTemplate"/>
       <property name="destination" ref="topicDest"/>
    </bean>

    <bean id="topicSubscriberService" class="jms.mq.spring.TopicSubscriberService">
       <property name="jmsTemplate" ref="jmsTopicTemplate"/>
       <property name="destination" ref="topicDest"/>
    </bean>
</beans>

评论

2 楼 dengwanchuan 2013-09-04  

pansonphy 写道

public class TopicPublisherService implements QueueProducerService和
public class TopicSubscriberService implements QueueConsumerService
这两个是不是有问题啊

是不是应该改成

public class TopicPublisherService extends QueueProducerService和
public class TopicSubscriberService extends QueueConsumerService

是有这个问题,避免程序复杂化,把implements 去掉了。TopicPublisherService 继承QueueProducerService也不好,他们是不同的模式，但是TopicPublisherService 和QueueProducerService可以抽象成一个父类.

1 楼 pansonphy 2013-09-03  

public class TopicPublisherService implements QueueProducerService和
public class TopicSubscriberService implements QueueConsumerService
这两个是不是有问题啊

是不是应该改成

public class TopicPublisherService extends QueueProducerService和
public class TopicSubscriberService extends QueueConsumerService