Spring与ActiveMQ整合

将spring与ActiveMQ整合。用spring管理对象的创建，以及spring提供的jmsTemplate中的方法去发送消息。采用spring的注解方式初始化对象。

spring下载地址:http://pan.baidu.com/s/1dFDusDF

activeMq jar包下载地址：http://pan.baidu.com/s/1pLygNv9

Spring与ActiveMQ整合需要的将相关jar包导入项目工程。



下面是消息发送的主函数：

package com.spring.jms;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class JmsTest {

public static void main(String[] args) {
ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext("beans.xml");
MessageService ms=(MessageService)(ctx.getBean("messageService"));
for (int i = 0; i < 10; i++) {
ms.sendMessage("消息"+i);
}

}
}

下面是封装的消息发送的服务，采用jmsTemplate.send()发送消息：

package com.spring.jms;

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;
@Component("messageService")
public class MessageService {

private JmsTemplate jmsTemplate;
private Destination mQueue;
@Resource
public void setmQueue(Destination mQueue) {
this.mQueue = mQueue;
}
@Resource
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public void sendMessage(final String message){

jmsTemplate.send(mQueue, new MessageCreator() {

@Override
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(message);
}
});
}

}

下面是消费者监听器，接收发送过来的消息：

package com.spring.jms;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

import org.springframework.stereotype.Component;
@Component("consumerMessageListener")
public class ConsumerMessageListener implements MessageListener {

@Override
public void onMessage(Message message) {
try {
System.out.println("接收消息"+((TextMessage)message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}

}

在封装的消息服务（MessageService ）代码中，注入的jmsTemplate,mQueue等对象都是通过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-2.5.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-2.5.xsd" >
<context:annotation-config />
<context:component-scan base-package="com.spring"/>
<!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->
<bean id="connectinFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616" />
</bean>
<!-- Spring Caching连接工厂 -->
<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="cachingConnectionFactory"
class="org.springframework.jms.connection.CachingConnectionFactory">
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="connectinFactory"></property>
<!-- Session缓存数量 -->
<property name="sessionCacheSize" value="10"></property>
</bean>
<!-- 配置消息发送目的地方式 -->
<!-- Queue队列：仅有一个订阅者会收到消息，消息一旦被处理就不会存在队列中 -->
<bean id="mQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg index="0" value="MessageQueue" />
</bean>
<!-- Spring JMS Template 配置JMS模版 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="cachingConnectionFactory" />
</bean>
<jms:listener-container container-type="default" connection-factory="cachingConnectionFactory" acknowledge="auto">
<jms:listener destination="MessageQueue" ref="consumerMessageListener"/>
</jms:listener-container>
</beans>

主要配置：

1. 配置ConnectionFactory

connectionFactory是Spring用于创建到JMS服务器链接的，Spring提供了多种connectionFactory，主要有SingleConnectionFactory和CachingConnectionFactory。

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

CachingConnectionFactory：继承了SingleConnectionFactory，新增了缓存功能，它可以缓存Session、MessageProducer和MessageConsumer。

Spring提供的ConnectionFactory只是Spring用于管理ConnectionFactory的，真正产生到JMS服务器链接的ConnectionFactory还得是由JMS服务厂商提供，并且需要把它注入到Spring提供的ConnectionFactory中。

<!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->
<bean id="connectinFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616" />
</bean>
<!-- Spring Caching连接工厂 -->
<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="cachingConnectionFactory"
class="org.springframework.jms.connection.CachingConnectionFactory">
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="connectinFactory"></property>
<!-- Session缓存数量 -->
<property name="sessionCacheSize" value="10"></property>
</bean>

2.配置生产者

生产者负责产生消息并发送到JMS服务器。Spring提供的JmsTemplate类来实现消息发送，定义JmsTemplate的时候需要注入一个Spring提供的ConnectionFactory对象。

<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="cachingConnectionFactory" />
</bean>

3.配置消息的发送目的地方式

两种方式，一种是queue，一种是topic，本例中采用queue模式

<bean id="mQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg index="0" value="MessageQueue" />
</bean>
<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="topic"/>
</bean>

4.配置消费者

1）通过配置MessageListenerContainer的时候往里面注入一个ConnectionFactory，这里是我们前面配置的“cachingConnectionFactory“，找到brokeUrl，这样才能知道监听的哪个服务器。

2）配置监听的目的地destination,这里是“MessageQueue”；

3）配置接收到消息以后进行消息处理的MessageListener，这里是“consumerMessageListener”

<jms:listener-container container-type="default" connection- factory="cachingConnectionFactory" acknowledge="auto">
<jms:listener destination="MessageQueue" ref="consumerMessageListener"/>
</jms:listener-container>

程序结果：