Spring整合JMS——基于ActiveMQ实现
Posted liguangsunls
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了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准备
配置ConnectionFactory
CachingConnectionFactory继承了SingleConnectionFactory。所以它拥有SingleConnectionFactory的全部功能，同
时它还新增了缓存功能。它能够缓存Session、MessageProducer和MessageConsumer。这里我们使用SingleConnectionFactory来作为演示样例。<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"/>
<!-- 真正能够产生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之后我们就须要配置生产者。
<!-- Spring提供的JMS工具类，它能够进行消息发送、接收等 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个connectionFactory相应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<property name="connectionFactory" ref="connectionFactory"/>
</bean> 在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>
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发送消息后，接下来就是消费者对指定目的地的消息进行消费了。
它非常好的平衡了对JMS提供者要求低、先进功能如事务參与和兼容Java EE环境
。定义处理消息的MessageListener
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();
}
}
}
<!--这个是队列目的地-->
<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> 这个时候完整的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));
}
}
} 
看，控制台已经进行了正确的输出，这说明我们的整合确实是已经成功了。
以上是关于Spring整合JMS——基于ActiveMQ实现的主要内容，如果未能解决你的问题，请参考以下文章
深入浅出JMS--Spring和ActiveMQ整合的完整实例