ActiveMQ集成Spring

一、环境准备

这里以 Maven 作为项目管理工具来创建工程。

项目目录结构：

1.1在`pom.xml`中添加依赖：

    <properties><spring-version>4.3.9.RELEASE</spring-version></properties><dependencies><dependency><groupId>org.springframework</groupId><artifactId>spring-context</artifactId><version>${spring-version}</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-jms</artifactId><version>${spring-version}</version></dependency><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-core</artifactId><version>5.7.0</version><!-- 排除 ActiveMQ 自身依赖的 Spring --><exclusions><exclusion><artifactId>spring-context</artifactId><groupId>org.springframework</groupId></exclusion></exclusions></dependency></dependencies>

1.2定义 Spring 的配置文件

以点对点或队列模型为例进行集成，关于发布者／订阅者模型会以注释的形式贴出来。为了减少篇幅，Spring 配置文件中的头文件不再贴出。

公共的 Spring 配置文件common.xml：

    <context:annotation-config/><!-- ActiveMQ 提供的ConnectionFactory --><bean id="activeMQConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"><!-- 配置 brokerURL，这里换为你自己开启 ActiveMQ 服务的地址--><property name="brokerURL" value="tcp://192.168.248.136:61616"/></bean><!-- 可以理解为 Spring 提供的连接池 --><bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"><property name="targetConnectionFactory" ref="activeMQConnectionFactory"/></bean><!-- 点对点或队列模型配置队列的目的地 --><bean id="activeMQQueue" class="org.apache.activemq.command.ActiveMQQueue"><constructor-arg name="name" value="spring-jms-queue"/></bean><!--发布者／订阅者模型配置主题的目的地 -->     <bean id="activeMQTopic" class="org.apache.activemq.command.ActiveMQTopic"><constructor-arg name="name" value="spring-jms-topic"/></bean>

生产者 Spring 配置文件producer.xml：

    <!-- 导入公共配置 --><import resource="common.xml"/><!-- 配置 JmsTemplate --><bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"><property name="connectionFactory" ref="connectionFactory"/></bean><!-- 把 ProducerServiceImpl 交给Spring IoC 容器管理--><bean class="com.jas.jms.producer.ProducerServiceImpl"/>

消费者 Spring 配置文件consumer.xml：

    <!-- 导入公共配置 --><import resource="common.xml"/><!-- 配置自定义消费者消息监听器 --><bean id="consumerMessageListener" class="com.jas.jms.consumer.ConsumerMessageListener"/><!-- 配置消息监听器的容器 --><bean id="container" class="org.springframework.jms.listener.DefaultMessageListenerContainer"><property name="connectionFactory" ref="connectionFactory"/><property name="destination" ref="activeMQQueue"/><!--配置发布者／订阅者模型的目的地<property name="destination" ref="activeMQTopic"/>--><property name="messageListener" ref="consumerMessageListener"/></bean>

二、编写对应的代码

2.1生产者

ProducerService接口

package com.jas.jms.producer;public interface ProducerService {/*** 生产者发送消息* @param message*/void sendMessage(String message);
}

ProducerServiceImpl实现类

package com.jas.jms.producer;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;import javax.annotation.Resource;
import javax.jms.*;public class ProducerServiceImpl implements ProducerService {@AutowiredJmsTemplate jmsTemplate;/*** 这里以 @Resource 方式注入目的地对象* 如果是发布者/订阅者模式，只选要修改 name 中的值为“activeMQTopic”即可*/@Resource(name = "activeMQQueue")Destination destination;@Overridepublic void sendMessage(final String message) {jmsTemplate.send(destination, new MessageCreator() {@Overridepublic Message createMessage(Session session) throws JMSException {TextMessage textMessage = session.createTextMessage(message);return textMessage;}});System.out.println("消息已发送：" + message);}
}

生产者调用类Producer

package com.jas.jms.producer;import org.springframework.context.support.ClassPathXmlApplicationContext;public class Producer {public static void main(String[] args) {ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer.xml");ProducerService producerService = context.getBean(ProducerService.class);// 生产者发送十次消息for (int i = 0; i < 10; i++) {producerService.sendMessage("test message：" + i);}// 关闭 IoC 容器context.close();}
}

2.2消费者

消费者消息监听ConsumerMessageListener

package com.jas.jms.consumer;import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;public class ConsumerMessageListener implements MessageListener {@Overridepublic void onMessage(Message message) {TextMessage textMessage = (TextMessage) message;try {System.out.println("接收已接收：" + textMessage.getText());} catch (JMSException e) {e.printStackTrace();}}
}

消费者调用类Producer

package com.jas.jms.consumer;import org.springframework.context.support.ClassPathXmlApplicationContext;public class Consumer {public static void main(String[] args) {ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("consumer.xml");}
}

2.3测试

首先开启两个消费者服务，在开启一个生产者服务。

消费者1输出：

消费者2输出：

2.4发布者／订阅者模型

在common.xml配置文件中已经配置了发布者／订阅者模式，只需要在consumer.xml配置文件中注释掉<property name="destination" ref="activeMQQueue"/>，并启用activeMQQueue。

还需要修改的一个地方是修改ProducerServiceImpl类中Destination destination;上面的注解为@Resource(name = "activeMQTopic")即可。

PS

代码 GitHub 地址：https://github.com/coderjas/activemq-study
什么是消息队列(Message queue)：https://blog.csdn.net/codejas/article/details/79930909

（完）