ActiveMQ与Spring整合

ActiveMQ 是Apache出品， 是最流行和最强大的开源消息总线。 同时完全支持 JMS 1.1和J2EE 1.4规范。
ActiveMQ 特性

支持多种编程语言和协议编写客户端。

在JMS客户端和消息代理完全支持企业集成模式

完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)。

对Spring的支持, ActiveMQ可以很容易内嵌到使用Spring的系统里面去,而且也支持Spring2.0的特性。

测试通过常见的J2EE服务器。如：Geronimo, JBoss 4, GlassFish, WebLogic。

支持多种传送协议：in-VM, TCP, SSL, NIO, UDP, JGroups,JXTA。

支持通过JDBC和journal提供高速的消息持久化。

从设计上保证了高性能的集群,客户端-服务器,点对点。

支持Ajax

支持与Axis的整合

可以很容易得调用内嵌JMS provider,进行测试。

通过网络搜索及个人理解，整理出ActiveMQ与Spring整合的文章。
相关jar包

activemq-core-5.5.1.jar
geronimo-j2ee-management_1.1_spec-1.0.1.jar
geronimo-jms_1.1_spec-1.1.1.jar
geronimo-jta_1.0.1B_spec-1.0.1.jar
kahadb-5.5.1.jar
xbean-spring-3.7.jar
commons-beanutils.jar
commons-codec.jar
commons-collections.jar
commons-fileupload.jar
commons-httpclient.jar
commons-io.jar
commons-lang.jar
commons-logging.jar
commons-validator.jar
dom4j-1.6.1.jar
javaee.jar
jsf-api.jar
jsf-impl.jar
jstl.jar
log4j-1.2.15.jar
slf4j-api-1.5.8.jar
slf4j-log4j12-1.5.8.jar
aspectjrt.jar
aspectjweaver.jar
spring.jar

spring.xml

<?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:p="http://www.springframework.org/schema/p"
xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:jee="http://www.springframework.org/schema/jee"
xmlns:amq="http://activemq.apache.org/schema/core"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-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/util http://www.springframework.org/schema/util/spring-util-2.5.xsd http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-2.5.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.5.0.xsd"> 
　　<!-- Annotation Config -->
　　<context:annotation-config/>

　　<!-- Compoent Scan -->
　　<context:component-scan base-package="com.uu.web.*"/>

　　<!-- Property Placeholder -->
　　<context:property-placeholder location="classpath:config.properties"/>

　　<!-- Aop Config -->
　　<aop:aspectj-autoproxy/>
　　<!--
　　<bean id="mysql-ds"  destroy-method="close">
　　　　<property name="driverClassName" value="${dirver}"/>
　　　　<property name="url" value="${url}"/>
　　　　<property name="username" value="${username}"/>
　　　　<property name="password" value="${password}"/>
　　</bean>
　　-->
　　<!--
　　　　使用spring的listenerContainer 消息用持久化保存，服务器重启不会丢失
　　　　也可以配置在${ACTIVEMQ_HOME}/conf/activemq.xml内
　　　　消息的保存方式文件持久化和数据库持久化
　　　　此配置是文件持久化
　　-->
　　<!-- Embedded ActiveMQ Broker -->
　　<amq:broker useJmx="false" persistent="true">
　　　　<amq:persistenceAdapter>
　　　　　　<amq:amqPersistenceAdapter directory="G:/amq"/>
　　　　　　<!-- 使用数据库持久化 -->
　　　　　　<!--<amq:jdbcPersistenceAdapter dataSource="#mysql-ds" />-->
　　　　</amq:persistenceAdapter>
　　　　<amq:transportConnectors>
　　　　　　<amq:transportConnector uri="tcp://localhost:61616" />
　　　　</amq:transportConnectors>
　　</amq:broker>

　　<!--ActiveMQ connectionFactory  -->
　　<amq:connectionFactory id="jmsConnectionFactory" brokerURL="vm://localhost" />

　　<!-- ActiveMQ destinations  -->
　　<!--
　　　　TOPIC：发布订阅消息
　　　　　　　　无状态，不保证每条消息被消费
　　　　　　　　只有监听该TOPIC地址才能收到消息并消费，否则该消息将会丢失
　　　　　　　　一对多的发布接受策略，可以同时消费多个消息
　　-->
　　<amq:topic name="TOPIC" physicalName="JMS-TOPIC" />

<!--
　　　　QUEUE：　点对点
　　　　　　　　　消息数据被持久化，每条消息都能被消费
　　　　　　　　　没有监听QUEUE地址也能被消费，数据不会丢失
　　　　　　　　　一对一的发布接受策略，保证数据完整
-->
<amq:queue name="QUEUE" physicalName="JMS-QUEUE" />

　　<!-- ConnectionFactory -->
　　<bean id="singleConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory" destroy-method="destroy">
　　　　<property name="targetConnectionFactory" ref="jmsConnectionFactory"/>
　　</bean>

　　<!-- 添加事务 -->
　　<bean id="jmsTransactionManager" class="org.springframework.jms.connection.JmsTransactionManager" >
　　　　<property name="connectionFactory" ref="singleConnectionFactory"/>
　　</bean>

　　<!--  Spring JmsTemplate config -->
　　<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
　　　　<!--  lets wrap in a pool to avoid creating a connection per send -->
　　　　<property name="connectionFactory" ref="singleConnectionFactory"/>
　　　　<!-- custom MessageConverter -->
　　　　<property name="messageConverter" ref="defaultMessageConverter" />
　　　　<property name="sessionTransacted" value="true"/>
　　</bean>

　　<!-- converter  -->
　　<bean id="defaultMessageConverter" class="com.uu.activemq.DefaultMessageConverter" />

　　<!-- 生产消息配置 -->
　　<!-- POJO which send Message uses  Spring JmsTemplate -->
　　<bean id="topicMessageProducer" class="com.uu.activemq.TopicMessageProducer">
　　　　<property name="template" ref="jmsTemplate" />
　　　　<property name="destination" ref="TOPIC" />
　　</bean>
　　<bean id="queueMessageProducer" class="com.uu.activemq.QueueMessageProducer">
　　　　<property name="template" ref="jmsTemplate" />
　　　　<property name="destination" ref="QUEUE" />
　　</bean>

　　<!-- 消费消息 配置 -->
　　<!--  Message Driven POJO (MDP) -->
　　<!-- consumer1 for topic -->
　　<bean id="topicConsumer" class="com.uu.activemq.TopicConsumer" />

　　<!-- consumer for queue -->
　　<bean id="queueConsumer" class="com.uu.activemq.QueueConsumer" />

　　<!-- Message Listener for  -->
　　<bean id="topicListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
　　　　<constructor-arg ref="topicConsumer" />
　　　　<!--  指定消费消息的方法 -->
　　　　<property name="defaultListenerMethod" value="receive" />
　　　　<!-- custom MessageConverter define -->
　　　　<property name="messageConverter" ref="defaultMessageConverter" />
　　</bean>
　　<bean id="queueListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
　　　　<constructor-arg ref="queueConsumer" />
　　　　<!--  指定消费消息的方法 -->
　　　　<property name="defaultListenerMethod" value="receive" />
　　　　<!-- custom MessageConverter define -->
　　　　<property name="messageConverter" ref="defaultMessageConverter" />
　　</bean>

　　<!--  listener container，MDP无需实现接口 -->
　　<bean id="topicListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
　　　　<property name="connectionFactory" ref="jmsConnectionFactory" />
　　　　<property name="destination" ref="TOPIC" />
　　　　<property name="messageListener" ref="topicListener" />
　　</bean>
　　<bean id="queueListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
　　　　<property name="connectionFactory" ref="jmsConnectionFactory" />
　　　　<property name="destination" ref="QUEUE" />
　　　　<property name="messageListener" ref="queueListener" />
　　　　<property name="transactionManager" ref="jmsTransactionManager"/>
　　　　<property name="sessionTransacted" value="true"/>
　　　　<property name="concurrentConsumers" value="5"/>
　　</bean>
</beans>

消息转换类
DefaultMessageConverter.java

1 package com.uu.activemq;
2
3 import java.io.ByteArrayInputStream;
4 import java.io.ByteArrayOutputStream;
5 import java.io.IOException;
6 import java.io.ObjectInputStream;
7 import java.io.ObjectOutputStream;
8 import java.util.HashMap;
9
10 import javax.jms.JMSException;
11 import javax.jms.Message;
12 import javax.jms.ObjectMessage;
13 import javax.jms.Session;
14
15 import org.apache.activemq.command.ActiveMQObjectMessage;
16 import org.apache.commons.logging.Log;
17 import org.apache.commons.logging.LogFactory;
18 import org.springframework.jms.support.converter.MessageConverter;
19
20 /**
21  * 消息转换类
22  */
23 @SuppressWarnings("unchecked")
24 public class DefaultMessageConverter implements MessageConverter {
25
26     private static final Log log = LogFactory.getLog(DefaultMessageConverter.class);
27
28     public Message toMessage(Object obj, Session session) throws JMSException {
29         if (log.isDebugEnabled()) {
30             log.debug("toMessage(Object, Session) - start");
31         }
32
33         // check Type
34         ActiveMQObjectMessage objMsg = (ActiveMQObjectMessage) session.createObjectMessage();
35         HashMap<String, byte[]> map = new HashMap<String, byte[]>();
36         try {
37             // POJO must implements Seralizable
38             ByteArrayOutputStream bos = new ByteArrayOutputStream();
39             ObjectOutputStream oos = new ObjectOutputStream(bos);
40             oos.writeObject(obj);
41             map.put("POJO", bos.toByteArray());
42             objMsg.setObjectProperty("Map", map);
43
44         } catch (IOException e) {
45             log.error("toMessage(Object, Session)", e);
46         }
47         return objMsg;
48     }
49
50
51     public Object fromMessage(Message msg) throws JMSException {
52         if (log.isDebugEnabled()) {
53             log.debug("fromMessage(Message) - start");
54         }
55
56         if (msg instanceof ObjectMessage) {
57             HashMap<String, byte[]> map = (HashMap<String, byte[]>) ((ObjectMessage) msg).getObjectProperty("Map");
58             try {
59                 // POJO must implements Seralizable
60                 ByteArrayInputStream bis = new ByteArrayInputStream(map.get("POJO"));
61                 ObjectInputStream ois = new ObjectInputStream(bis);
62                 Object returnObject = ois.readObject();
63                 return returnObject;
64             } catch (IOException e) {
65                 log.error("fromMessage(Message)", e);
66
67             } catch (ClassNotFoundException e) {
68                 log.error("fromMessage(Message)", e);
69             }
70
71             return null;
72         } else {
73             throw new JMSException("Msg:[" + msg + "] is not Map");
74         }
75     }
76 }

接受Queue方式消费
QueueConsumer.java

1 package com.uu.activemq;
2
3 import java.util.Map;
4 import java.util.Set;
5
6 public class QueueConsumer
7 {
8     public void receive(Map<String, Object> message)
9     {
10         Set<String> set = message.keySet();
11         String str = "";
12         for(String key : set)
13         {
14             str += key + "_" + message.get(key) + ":Queue";
15         }
16         System.out.println(str);
17     }
18 }

接受Topic方式消费
TopicConsumer.java

1 package com.uu.activemq;
2
3 import java.util.Map;
4 import java.util.Set;
5 import java.util.concurrent.ExecutorService;
6 import java.util.concurrent.Executors;
7
8 /**
9  * 创建10个线程池
10  * 1.使用线程异步消息处理
11  * 2.不使用线程，那么消息等待上一个消息处理完成后才继续
12  * 注:如果担心据同步问题，那么使用第2种方法
13  */
14 public class TopicConsumer
15 {
16     protected static ExecutorService exec = Executors.newFixedThreadPool(10);
17
18     public void receive(Map<String, Object> message)
19     {
20         /*exec.submit(new Runnable(){
21
22             public void run()
23             {
24
25             }
26         });*/
27         Set<String> set = message.keySet();
28         String str = "";
29         for(String key : set)
30         {
31             str += key + "_" + message.get(key) + ":Topic";
32         }
33         System.out.println(str);
34     }
35 }

生产Queue方式的消息
QueueMessageProducer.java

1 package com.uu.activemq;
2
3 import java.util.Map;
4
5 import javax.jms.Queue;
6
7 import org.springframework.jms.core.JmsTemplate;
8
9 public class QueueMessageProducer {
10
11     private JmsTemplate template;
12
13     private Queue destination;
14
15     public void setTemplate(JmsTemplate template) {
16         this.template = template;
17     }
18
19     public void setDestination(Queue destination) {
20         this.destination = destination;
21     }
22
23     public void send(Map<String,Object> message) {
24         template.convertAndSend(this.destination, message);
25     }
26
27 }

生产Topic方式的消息
TopicMessageProducer.java

1 package com.uu.activemq;
2
3 import java.util.Map;
4
5 import javax.jms.Topic;
6
7 import org.springframework.jms.core.JmsTemplate;
8
9 public class TopicMessageProducer {
10
11     private JmsTemplate template;
12
13     private Topic destination;
14
15     public void setTemplate(JmsTemplate template) {
16         this.template = template;
17     }
18
19     public void setDestination(Topic destination) {
20         this.destination = destination;
21     }
22
23     public void send(Map<String,Object> message) {
24         template.convertAndSend(this.destination, message);
25     }
26 }

写一个测试类

1 package com.uu.activemq;
2
3 import java.util.LinkedHashMap;
4 import java.util.Map;
5
6 import org.springframework.context.ApplicationContext;
7 import org.springframework.context.support.FileSystemXmlApplicationContext;
8
9 /**
10  * 解决耗时的数据操作
11  * 发送消息不等待返回，继续执行
12  */
13 public class MainTest
14 {
15     public static void main(String[] args)
16     {
17         ApplicationContext wac = new FileSystemXmlApplicationContext("classpath:spring.xml");
18
19         TopicMessageProducer topicMessageProducer  = (TopicMessageProducer) wac.getBean("topicMessageProducer");
20
21         QueueMessageProducer queueMessageProducer  = (QueueMessageProducer) wac.getBean("queueMessageProducer");
22
23         Map<String, Object> message = new LinkedHashMap<String, Object>();
24         message.put("test", "ActiveMQ");
25
26         queueMessageProducer.send(message);
27         topicMessageProducer.send(message);
28         System.out.println("完成");
29     }
30 }

结果先输出完成然后输出传递的参数

完成
test_ActiveMQ:Queue
test_ActiveMQ:Topic