Spring整合JMS(一)——基于ActiveMQ实现

JMS简单介绍

JMS的全称是Java Message Service。即Java消息服务。

它主要用于在生产者和消费者之间进行消息传递，生产者负责产生消息。而消费者负责接收消息。

把它应用到实际的业务需求中的话我们能够在特定的时候利用生产者生成一消息，并进行发送，相应的消费者在接收到相应的消息后去完毕相应的业务逻辑。

对于消息的传递有两种类型，一种是点对点的。即一个生产者和一个消费者一一相应。还有一种是公布/订阅模式，即一个生产者产生消息并进行发送后，能够由多个消费者进行接收。

Spring整合JMS

对JMS做了一个简要介绍之后，接下来就讲一下Spring整合JMS的详细过程。

JMS仅仅是一个标准，真正在使用它的时候我们须要有它的详细实现,这里我们就使用Apache的activeMQ来作为它的实现。所使用的依赖利用Maven来进行管理。详细依赖例如以下：

<!-- jms -->
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-jms_1.1_spec</artifactId>
<version>${geronimo.version}</version>
</dependency>
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-jta_1.1_spec</artifactId>
<version>${geronimo.version}</version>
</dependency>

<!-- activeMQ -->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-core</artifactId>
<version>${activemq.version}</version>
<exclusions>
<exclusion>
<groupId>org.apache.activemq.protobuf</groupId>
<artifactId>activemq-protobuf</artifactId>
</exclusion>
<exclusion>
<groupId>org.springframework.osgi</groupId>
<artifactId>spring-osgi-core</artifactId>
</exclusion>
<exclusion>
<groupId>org.osgi</groupId>
<artifactId>org.osgi.core</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
<version>${activemq.version}</version>
</dependency>

<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>${spring.version}</version>
</dependency>

ActiveMQ准备

既然是使用的apache的activeMQ作为JMS的实现，那么首先我们应该到apache官网上下载activeMQ（http://activemq.apache.org/download.html），进行解压后执行其bin文件夹以下的activemq.bat文件启动activeMQ。

配置ConnectionFactory

ConnectionFactory是用于产生到JMSserver的链接的，Spring为我们提供了多个ConnectionFactory。有SingleConnectionFactory和CachingConnectionFactory。SingleConnectionFactory对于建立JMSserver链接的请求会一直返回同一个链接，而且会忽略Connection的close方法调用。

CachingConnectionFactory继承了SingleConnectionFactory。所以它拥有SingleConnectionFactory的全部功能，同时它还新增了缓存功能。它能够缓存Session、MessageProducer和MessageConsumer。这里我们使用SingleConnectionFactory来作为演示样例。

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

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

<!-- 真正能够产生Connection的ConnectionFactory。由相应的 JMS服务厂商提供-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
</bean>

<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<!-- 目标ConnectionFactory相应真实的能够产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="targetConnectionFactory"/>
</bean>

配置生产者配置好ConnectionFactory之后我们就须要配置生产者。

生产者负责产生消息并发送到JMSserver。这通常相应的是我们的一个业务逻辑服务实现类。

可是我们的服务实现类是怎么进行消息的发送的呢？这一般是利用Spring为我们提供的JmsTemplate类来实现的，所以配置生产者事实上最核心的就是配置进行消息发送的JmsTemplate。对于消息发送者而言，它在发送消息的时候要知道自己该往哪里发。为此，我们在定义JmsTemplate的时候须要往里面注入一个Spring提供的ConnectionFactory对象。

<!-- Spring提供的JMS工具类，它能够进行消息发送、接收等 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个connectionFactory相应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<property name="connectionFactory" ref="connectionFactory"/>
</bean>

在真正利用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>

如果我们定义了一个ProducerService，里面有一个向Destination发送纯文本消息的方法sendMessage，那么我们的代码就大概是这个样子：

package com.somnus.jms.service.impl;

import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

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

import com.somnus.jms.service.ProducerService;

@Component
public class ProducerServiceImpl implements ProducerService {

private JmsTemplate jmsTemplate;

public void sendMessage(Destination destination, final String message) {
System.out.println("---------------生产者发送消息-----------------");
System.out.println("---------------生产者发了一个消息：" + message);
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(message);
}
});
}

public JmsTemplate getJmsTemplate() {
returnjmsTemplate;
}

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

}

我们能够看到在sendMessage方法体里面我们是通过jmsTemplate来发送消息到相应的Destination的。

到此，我们生成一个简单的文本消息并把它发送到指定目的地Destination的生产者就配置好了。

配置消费者生产者往指定目的地Destination发送消息后，接下来就是消费者对指定目的地的消息进行消费了。

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

Spring一共为我们提供了两种类型的MessageListenerContainer。SimpleMessageListenerContainer和DefaultMessageListenerContainer。SimpleMessageListenerContainer会在一開始的时候就创建一个会话session和消费者Consumer，而且会使用标准的JMS MessageConsumer.setMessageListener()方法注冊监听器让JMS提供者调用监听器的回调函数。它不会动态的适应执行时须要和參与外部的事务管理。兼容性方面。它很接近于独立的JMS规范，但一般不兼容Java
EE的JMS限制。大多数情况下我们还是使用的DefaultMessageListenerContainer。跟SimpleMessageListenerContainer相比，DefaultMessageListenerContainer会动态的适应执行时须要。而且可以參与外部的事务管理。

它非常好的平衡了对JMS提供者要求低、先进功能如事务參与和兼容Java
EE环境。

定义处理消息的MessageListener

要定义处理消息的MessageListener我们仅仅须要实现JMS规范中的MessageListener接口就能够了。

MessageListener接口中仅仅有一个方法onMessage方法，当接收到消息的时候会自己主动调用该方法。

package com.somnus.jms.listener;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

public class ConsumerMessageListener implements MessageListener {

public void onMessage(Message message) {
//这里我们知道生产者发送的就是一个纯文本消息，所以这里能够直接进行强制转换。或者直接把onMessage方法的參数改成Message的子类TextMessage
TextMessage textMsg = (TextMessage) message;
System.out.println("接收到一个纯文本消息。");
try {
System.out.println("消息内容是：" + textMsg.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}

}

有了MessageListener之后我们就能够在Spring的配置文件里配置一个消息监听容器了。

<!--这个是队列目的地-->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg>
<value>queue</value>
</constructor-arg>
</bean>
<!-- 消息监听器 -->
<bean id="consumerMessageListener" class="com.somnus.jms.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.somnus" />

<!-- Spring提供的JMS工具类，它能够进行消息发送、接收等 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个connectionFactory相应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<property name="connectionFactory" ref="connectionFactory"/>
</bean>

<!-- 真正能够产生Connection的ConnectionFactory，由相应的 JMS服务厂商提供-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
</bean>

<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<!-- 目标ConnectionFactory相应真实的能够产生JMS Connection的ConnectionFactory -->
<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.somnus.jms.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>

接着我们来測试一下。看看我们的整合是否真的成功了，測试代码例如以下：

package com.somnus.jms.test;

import javax.jms.Destination;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.somnus.jms.service.ProducerService;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration("/applicationContext.xml")
public class ProducerConsumerTest {

@Autowired
private ProducerService producerService;
@Autowired
@Qualifier("queueDestination")
private Destination destination;

@Test
public void testSend() {
for (int i=0; i<2; i++) {
producerService.sendMessage(destination, "你好，生产者！这是消息：" + (i+1));
}
}

}

在上面的測试代码中我们利用生产者发送了两个消息，正常来说，消费者应该能够接收到这两个消息。执行測试代码后控制台输出例如以下：



看，控制台已经进行了正确的输出，这说明我们的整合确实是已经成功了。