ActiveMQ集成spring

<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.15.0</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.8.1</version>
</dependency>

<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>4.3.9.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
<version>4.3.9.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>4.3.9.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>4.3.9.RELEASE</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-core</artifactId>
<version>5.7.0</version>
<exclusions>
<exclusion>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
</exclusion>
</exclusions>
</dependency>

2.编写一个 spring-activemq-common.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:context="http://www.springframework.org/schema/context"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> <context:component-scan base-package="com.spf.activemq.spring" /><!--ActiveMQ提供的ConnectionFactory--><bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"><property name="brokerURL" value="tcp://192.168.1.125:61616"/></bean><!--Spring jms 提供的连接池--><bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"><property name="targetConnectionFactory"  ref="targetConnectionFactory"/></bean><!--一个队列的目的地 点对点--><bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue"><constructor-arg value="spf-queue" /></bean><!--一个主题的目的地 点对点--><bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic"><constructor-arg value="spf-topic" /></bean></beans>

3.编写一个 spring-activemq-producer.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:context="http://www.springframework.org/schema/context"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> <context:component-scan base-package="com.spf.activemq.spring" /><import resource="spring-activemq-common.xml"/><!--配置消息发送模板--><bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"><property name="connectionFactory" ref="connectionFactory"/><property name="defaultDestination" ref="queueDestination"/></bean><!--队列发送--><bean class="com.spf.activemq.spring.producer.ProducerQueueServiceImpl"><property name="jmsTemplate" ref="jmsTemplate"/></bean><!--主题发送--><bean class="com.spf.activemq.spring.producer.ProducerTopicServiceImpl"><property name="topicJmsTemplate" ref="jmsTemplate"/></bean></beans>

<?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"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> <context:component-scan base-package="com.spf.activemq.spring" /><import resource="spring-activemq-common.xml"/><!--配置消息监听器--><bean id="queueMessageListener" class="com.spf.activemq.spring.customer.MessageListener"></bean><!--配置消息容器--><bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"><property name="connectionFactory" ref="connectionFactory"/><property name="destination" ref="queueDestination"/><property name="messageListener" ref="queueMessageListener"/></bean><bean id="topicJmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"><property name="connectionFactory" ref="connectionFactory"/><property name="destination" ref="topicDestination"/><property name="messageListener" ref="queueMessageListener"/></bean></beans>

5.编写一个主题模式和队列模式的service和实现，测试方法

5.1 队列的service和实现，测试方法

public interface ProducerQueueService {void sendMessage(String queueName, String msg);void sendMessage(String msg);}

实现：

public class ProducerQueueServiceImpl implements ProducerQueueService {private JmsTemplate jmsTemplate;/*** 向指定队列名发送消息* @param queueName 队列名* @param msg 消息*/public void sendMessage(String queueName, final String msg) {//使用 jmsTemplate 发送消息jmsTemplate.send(queueName, new MessageCreator() {//创建一条消息public Message createMessage(Session session) throws JMSException {TextMessage textMessage = session.createTextMessage(msg);return textMessage;}});System.out.println("向"+queueName+"队列发送消息："+ msg);}/*** 向默认队列发送消息* @param msg 消息*/public void sendMessage(final String msg) {String queueName = jmsTemplate.getDefaultDestination().toString();//使用 jmsTemplate 发送消息jmsTemplate.send(queueName, new MessageCreator() {//创建一条消息public Message createMessage(Session session) throws JMSException {TextMessage textMessage = session.createTextMessage(msg);return textMessage;}});System.out.println("向"+queueName+"队列发送消息："+ msg);}public JmsTemplate getJmsTemplate() {return jmsTemplate;}public void setJmsTemplate(JmsTemplate jmsTemplate) {this.jmsTemplate = jmsTemplate;}}

测试方法：

public class AppQueueProducer {public static void main(String[] args) {ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext("spring-activemq-producer.xml");ProducerQueueService producerService = applicationContext.getBean(ProducerQueueService.class);for (int i=0;i<100;i++){producerService.sendMessage("spf-queue","queue test"+i);}applicationContext.close();}}

5.1 主题的service和实现，测试方法

service:

public interface ProducerTopicService {void sendMessage(String queueName, String msg);void sendMessage(String msg);}

public class ProducerTopicServiceImpl implements ProducerTopicService {private JmsTemplate topicJmsTemplate;public void sendMessage(String topicName, final String msg) {topicJmsTemplate.setPubSubDomain(true);//使用 jmsTemplate 发送消息topicJmsTemplate.send(topicName, new MessageCreator() {//创建一条消息public Message createMessage(Session session) throws JMSException {TextMessage textMessage = session.createTextMessage(msg);return textMessage;}});System.out.println("向"+topicName+"主题发送消息："+ msg);}public void sendMessage(final String msg) {topicJmsTemplate.setPubSubDomain(true);String topicName = topicJmsTemplate.getDefaultDestination().toString();//使用 jmsTemplate 发送消息topicJmsTemplate.send(topicName, new MessageCreator() {//创建一条消息public Message createMessage(Session session) throws JMSException {TextMessage textMessage = session.createTextMessage(msg);return textMessage;}});System.out.println("向"+topicName+"主题发送消息："+ msg);}public void setTopicJmsTemplate(JmsTemplate topicJmsTemplate) {this.topicJmsTemplate = topicJmsTemplate;}public JmsTemplate getTopicJmsTemplate() {return topicJmsTemplate;}}

public class AppTopicProducer {public static void main(String[] args) {ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext("spring-activemq-producer.xml");ProducerTopicService producerService = applicationContext.getBean(ProducerTopicService.class);for (int i=0;i<100;i++){producerService.sendMessage("spf-topic","topic test"+i);}applicationContext.close();}}

public class MessageListener implements javax.jms.MessageListener {public void onMessage(Message message) {TextMessage textMessage = (TextMessage) message;try {System.out.println("接收消息："+textMessage.getText());} catch (JMSException e) {e.printStackTrace();}}}

7.写一个类来加载 spring-activemq-customer.xml 配置文件，启动监听器

8.测试队列模式发送 启动两个Customer看效果