spring整合使用activemq

activemq简单的demo这里就不做演示了，今天介绍一下如何利用spring整合activemq，也是实际工作中涉及到的，希望对各位伙伴有所帮助，

1、安装activemq，为演示方便，我已经提前在linux中安装了activemq，安装过程比较简单，应该说activemq是软件中算是最简单的一种了，只需要解压，进入bin目录启动即可，不需做其他的配置；

只需要一条命令，即可启动，注意需要把activemq的端口配置到防护墙的文件中，然后重启防火墙，否则浏览器直接访问是访问不到的，

成功启动后，浏览器输入：http://192.168.111.130:8161/admin，可进入管控台查看，

在这个页面可以进行相关的配置，具体可参考官网说明，为后面的使用，我这里提前创建一个队列名称，test_queue,后面可以直接拿来使用；

接下来是项目和代码的部分；

2、项目结构，

为模拟真实的场景，这里我创建了两个maven项目，首先看producer的部分，

pom.xml文件，

<dependencies><dependency><groupId>org.springframework</groupId><artifactId>spring-core</artifactId><version>4.0.0.RELEASE</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-context</artifactId><version>4.0.0.RELEASE</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-orm</artifactId><version>4.0.0.RELEASE</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-web</artifactId><version>4.0.0.RELEASE</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-jdbc</artifactId><version>4.0.0.RELEASE</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-webmvc</artifactId><version>4.0.0.RELEASE</version></dependency><!-- spring aop --><dependency><groupId>org.springframework</groupId><artifactId>spring-aspects</artifactId><version>4.0.0.RELEASE</version></dependency><dependency><groupId>com.mchange</groupId><artifactId>c3p0</artifactId><version>0.9.2</version></dependency><dependency><groupId>cglib</groupId><artifactId>cglib</artifactId><version>2.2</version></dependency><dependency><groupId>org.aspectj</groupId><artifactId>aspectjweaver</artifactId><version>1.6.8</version></dependency><!-- Spring整合MyBatis --><dependency><groupId>org.mybatis</groupId><artifactId>mybatis</artifactId><version>3.2.8</version></dependency><dependency><groupId>org.mybatis</groupId><artifactId>mybatis-spring</artifactId><version>1.2.2</version></dependency><!-- 控制日志输出：结合log4j --><dependency><groupId>log4j</groupId><artifactId>log4j</artifactId><version>1.2.17</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.7</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>5.1.37</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-context-support</artifactId><version>4.0.0.RELEASE</version></dependency><dependency><groupId>commons-collections</groupId><artifactId>commons-collections</artifactId><version>3.1</version></dependency><dependency><groupId>commons-codec</groupId><artifactId>commons-codec</artifactId><version>1.9</version></dependency><dependency><groupId>commons-io</groupId><artifactId>commons-io</artifactId><version>2.4</version></dependency><dependency><groupId>org.codehaus.jackson</groupId><artifactId>jackson-mapper-asl</artifactId><version>1.9.2</version></dependency><dependency><groupId>commons-fileupload</groupId><artifactId>commons-fileupload</artifactId><version>1.3.1</version></dependency><dependency><groupId>javax.servlet</groupId><artifactId>servlet-api</artifactId><version>2.5</version></dependency><dependency><groupId>javax.servlet.jsp</groupId><artifactId>jsp-api</artifactId><version>2.1.3-b06</version></dependency><dependency><groupId>jstl</groupId><artifactId>jstl</artifactId><version>1.2</version></dependency><!-- spring单元测试 --><dependency><groupId>org.springframework</groupId><artifactId>spring-test</artifactId><version>4.3.7.RELEASE</version><scope>test</scope></dependency><!-- 格式化对象，方便输出日志 --><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.1.41</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>druid</artifactId><version>1.0.9</version></dependency><dependency><groupId>org.aspectj</groupId><artifactId>aspectjweaver</artifactId><version>1.8.8</version></dependency><!--rabbitmq依赖 --><!-- <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>3.6.0</version> </dependency> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>1.6.5.RELEASE</version> </dependency> --><dependency><groupId>org.springframework</groupId><artifactId>spring-jms</artifactId><version>4.0.0.RELEASE</version></dependency><!--active mq start --><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-core</artifactId><version>5.7.0</version></dependency><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-pool</artifactId><version>5.13.3</version></dependency><!--active mq end --></dependencies>

大家可以根据自己需要，把主要的配置文件引入即可，整合spring的时候，倒数第三个是必须的；

log4j.properties老套路，这里不多解释，请看mq.properties，这个里面是连接linux中的activemq的主要配置信息，

## MQ
mq.brokerURL=tcp\://192.168.111.130\:61616
mq.userName=admin
mq.password=admin
mq.pool.maxConnections=10
#queueName
queueName=test_queue

spring-mq.xml，这个配置文件里，配置的项如ConnectionFactory，池化连接，activeMqJmsTemplate（队列模板）等，主要是方便项目代码中的连接使用，方便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:p="http://www.springframework.org/schema/p"xmlns:context="http://www.springframework.org/schema/context"xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"xsi:schemaLocation="http://www.springframework.org/schema/beans  http://www.springframework.org/schema/beans/spring-beans-3.2.xsd  http://www.springframework.org/schema/aop   http://www.springframework.org/schema/aop/spring-aop-3.2.xsd  http://www.springframework.org/schema/tx  http://www.springframework.org/schema/tx/spring-tx-3.2.xsd  http://www.springframework.org/schema/context  http://www.springframework.org/schema/context/spring-context-3.2.xsd"default-autowire="byName" default-lazy-init="false"><!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供 --><bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"><!-- ActiveMQ服务地址 --><property name="brokerURL" value="${mq.brokerURL}" /><property name="userName" value="${mq.userName}"></property><property name="password" value="${mq.password}"></property> </bean><!-- ActiveMQ为我们提供了一个PooledConnectionFactory，通过往里面注入一个ActiveMQConnectionFactory可以用来将Connection、Session和MessageProducer池化，这样可以大大的减少我们的资源消耗。要依赖于 activemq-pool包--><bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory"><property name="connectionFactory" ref="targetConnectionFactory" /><property name="maxConnections" value="${mq.pool.maxConnections}" /></bean><!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory --><bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"><!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory --><property name="targetConnectionFactory" ref="pooledConnectionFactory" /></bean><!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 --><!-- 队列模板 --><bean id="activeMqJmsTemplate" class="org.springframework.jms.core.JmsTemplate">  <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  <property name="connectionFactory" ref="connectionFactory"/>  <property name="defaultDestinationName" value="${queueName}"></property></bean> </beans>

接下来是和spring配置文件整合在一起，请看spring-context.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:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop"xmlns:tx="http://www.springframework.org/schema/tx"xsi:schemaLocation="http://www.springframework.org/schema/beans  http://www.springframework.org/schema/beans/spring-beans-3.2.xsd  http://www.springframework.org/schema/aop   http://www.springframework.org/schema/aop/spring-aop-3.2.xsd  http://www.springframework.org/schema/tx  http://www.springframework.org/schema/tx/spring-tx-3.2.xsd  http://www.springframework.org/schema/context  http://www.springframework.org/schema/context/spring-context-3.2.xsd"default-autowire="byName" default-lazy-init="false"><!-- 采用注释的方式配置bean --><context:annotation-config /><!-- 配置要扫描的包 --><context:component-scan base-package="wusc.edu.demo" /><!-- 读入配置属性文件 --><context:property-placeholder location="classpath:mq.properties" /><!-- proxy-target-class默认"false",更改为"ture"使用CGLib动态代理 --><aop:aspectj-autoproxy proxy-target-class="true" />    <import resource="spring-mq.xml" /></beans>

里面主要包括，引入mq的配置文件，扫描包，加载spring-mq的配置文件，比较简单，相信大家一眼就能看明白；

下面是具体的代码，这里为了模拟生产者，使用简单的点对点的方式进行消息的发送；
MailParam是个实体类，可以作为一个对象进行传输，

public class MailParam {/** 发件人 **/private String from;/** 收件人 **/private String to;/** 主题 **/private String subject;/** 邮件内容 **/private String content;public MailParam() {}public MailParam(String to, String subject, String content) {this.to = to;this.subject = subject;this.content = content;}public String getFrom() {return from;}public void setFrom(String from) {this.from = from;}public String getTo() {return to;}public void setTo(String to) {this.to = to;}public String getSubject() {return subject;}public void setSubject(String subject) {this.subject = subject;}public String getContent() {return content;}public void setContent(String content) {this.content = content;}
}

MQProducer：

@Service("mqProducer")
public class MQProducer {@Autowiredprivate JmsTemplate activeMqJmsTemplate;public void sendMessage(final MailParam mail){activeMqJmsTemplate.send(new MessageCreator() {public Message createMessage(Session session) throws JMSException {return session.createTextMessage(JSONObject.toJSONString(mail));}});}}

在test目录下创建一个测试类MQProducerTest，

public class MQProducerTest {private static final Log log = LogFactory.getLog(MQProducerTest.class);public static void main(String[] args) {try {ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("classpath:spring/spring-context.xml");context.start();MQProducer mqProducer = (MQProducer) context.getBean("mqProducer");MailParam mail = new MailParam();mail.setTo("2639378441@qq.com");mail.setSubject("ActiveMQ测试");mail.setContent("通过ActiveMQ异步发送邮件！");mqProducer.sendMessage(mail);} catch (Exception e) {log.error("==>MQ context start error:", e);System.exit(0);}finally{log.info("===>System.exit");System.exit(0);}}}

再看消费端，

pom文件和producer基本一致，不再罗列，配置文件部门也差不多，从项目结构可以看出来，在和spring进行整合的时候，消费端常常需要设定一个监听器，进行异步的监听消费，这个和rocketmq或rabbitmq思想类似，log4j.properties不再罗列，mq.properties如下，

## MQ
mq.brokerURL=tcp\://192.168.111.130\:61616
mq.userName=admin
mq.password=admin
mq.pool.maxConnections=10
#queueName
queueName=test_queue

注意到，生产者个消费端都采用了连接池，用于管理并发的连接数，方便管理和控制，这一点请注意；

spring-mq.xml，这个和producer中类似，多加了消息监听器的配置部分，

<?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:context="http://www.springframework.org/schema/context"xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"xsi:schemaLocation="http://www.springframework.org/schema/beans  http://www.springframework.org/schema/beans/spring-beans-3.2.xsd  http://www.springframework.org/schema/aop   http://www.springframework.org/schema/aop/spring-aop-3.2.xsd  http://www.springframework.org/schema/tx  http://www.springframework.org/schema/tx/spring-tx-3.2.xsd  http://www.springframework.org/schema/context  http://www.springframework.org/schema/context/spring-context-3.2.xsd"default-autowire="byName" default-lazy-init="false"><!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供 --><bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"><!-- ActiveMQ服务地址 --><property name="brokerURL" value="${mq.brokerURL}" /><property name="userName" value="${mq.userName}"></property><property name="password" value="${mq.password}"></property> </bean><!-- ActiveMQ为我们提供了一个PooledConnectionFactory，通过往里面注入一个ActiveMQConnectionFactory可以用来将Connection、Session和MessageProducer池化，这样可以大大的减少我们的资源消耗。要依赖于 activemq-pool包--><bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory"><property name="connectionFactory" ref="targetConnectionFactory" /><property name="maxConnections" value="${mq.pool.maxConnections}" /></bean><!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory --><bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"><!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory --><property name="targetConnectionFactory" ref="pooledConnectionFactory" /></bean><!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 --><!-- 队列模板 --><bean id="activeMqJmsTemplate" class="org.springframework.jms.core.JmsTemplate">  <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  <property name="connectionFactory" ref="connectionFactory"/>  <property name="defaultDestinationName" value="${queueName}"></property></bean> <!--这个是sessionAwareQueue目的地 --><bean id="sessionAwareQueue" class="org.apache.activemq.command.ActiveMQQueue"><constructor-arg><value>${queueName}</value></constructor-arg></bean><!-- 可以获取session的MessageListener --><bean id="consumerSessionAwareMessageListener" class="wusc.edu.demo.mqtest.listener.ConsumerSessionAwareMessageListener"></bean><bean id="sessionAwareListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"><property name="connectionFactory" ref="connectionFactory" /><property name="destination" ref="sessionAwareQueue" /><property name="messageListener" ref="consumerSessionAwareMessageListener" /></bean></beans>

spring-context.xml，用于引入外部配置文件并整合activemq的配置文件，

<?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:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop"xmlns:tx="http://www.springframework.org/schema/tx"xsi:schemaLocation="http://www.springframework.org/schema/beans  http://www.springframework.org/schema/beans/spring-beans-3.2.xsd  http://www.springframework.org/schema/aop   http://www.springframework.org/schema/aop/spring-aop-3.2.xsd  http://www.springframework.org/schema/tx  http://www.springframework.org/schema/tx/spring-tx-3.2.xsd  http://www.springframework.org/schema/context  http://www.springframework.org/schema/context/spring-context-3.2.xsd"default-autowire="byName" default-lazy-init="false"><!-- 采用注释的方式配置bean --><context:annotation-config /><!-- 配置要扫描的包 --><context:component-scan base-package="wusc.edu.demo" /><!-- 读入配置属性文件 --><context:property-placeholder location="classpath:mq.properties,classpath:mail.properties" /><!-- proxy-target-class默认"false",更改为"ture"使用CGLib动态代理 --><aop:aspectj-autoproxy proxy-target-class="true" />  <import resource="spring-mq.xml" /><import resource="spring-mail.xml" /></beans>

接下来，请看消息监听部分的代码，ConsumerSessionAwareMessageListener，这个类需要实现一个接口，SessionAwareMessageListener，

@Component
public class ConsumerSessionAwareMessageListener implements SessionAwareMessageListener<Message> {private static final Log log = LogFactory.getLog(ConsumerSessionAwareMessageListener.class);@Autowiredprivate JmsTemplate activeMqJmsTemplate;@Autowiredprivate Destination sessionAwareQueue;@Autowiredprivate MailBiz bailBiz;public synchronized void onMessage(Message message, Session session) {try {ActiveMQTextMessage msg = (ActiveMQTextMessage) message;final String ms = msg.getText();log.info("==>receive message:" + ms);MailParam mailParam = JSONObject.parseObject(ms, MailParam.class);// 转换成相应的对象if (mailParam == null) {return;}try {System.out.println("收到的消息内容是==============>>>>>>>>>>>"+mailParam.getContent());//bailBiz.mailSend(mailParam);} catch (Exception e) {// 发送异常，重新放回队列
//              activeMqJmsTemplate.send(sessionAwareQueue, new MessageCreator() {
//                  public Message createMessage(Session session) throws JMSException {
//                      return session.createTextMessage(ms);
//                  }
//              });log.error("==>MailException:", e);}} catch (Exception e) {log.error("==>", e);}}}

最后是消费端的启动测试类，MQConsumer，

public class MQConsumer {private static final Log log = LogFactory.getLog(MQConsumer.class);public static void main(String[] args) {try {ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("classpath:spring/spring-context.xml");context.start();System.out.println("consumer is starting.....");} catch (Exception e) {log.error("==>MQ context start error:", e);System.exit(0);}}
}

接下来就可以启动两个测试类，进行测试啦，首先启动消费端的代码，

此时，消费者成功启动，消息监听器启动，等待接收消息，再启动生产者，

此时，生产者也启动成功，并且成功发送一条消息到消息队列中，此时再看生产者端控制台输出的结果，

再看管控台，

可以看到，消费端成功收到来自生产者的消息，到此，spring整合activemq结束，需要代码的伙伴们可以下载：https://download.csdn.net/download/zhangcongyi420/10687136

大家还可以继续在此基础上尝试在ssm项目中整合activemq，原理基本类似，最后感谢大家观看！！！

spring整合使用activemq相关推荐

Spring整合ActiveMQ完成消息队列MQ编程
<–start–> 第一步:新建一个maven,将工程命名为activeMQ_spring.在pom.xml文件中导入相关jar包. ①spring开发和测试相关的jar包: spring ...
ActiveMQ与spring整合
2019独角兽企业重金招聘Python工程师标准>>> 1 生产者第一步:引用相关的jar包. <dependency> <groupId>org.spri ...
Spring整合JMS——基于ActiveMQ实现（一）
Spring整合JMS--基于ActiveMQ实现(一) 1.1 JMS简介 JMS的全称是Java Message Service,即Java消息服务.它主要用于在生产者和消费者之间进行消息 ...
activimq java集成_Java消息队列-Spring整合ActiveMq
1.概述首先和大家一起回顾一下Java 消息服务,在我之前的博客<Java消息队列-JMS概述>中,我为大家分析了: 消息服务:一个中间件,用于解决两个活多个程序之间的耦合,底层由Jav ...
ActiveMQ —— Spring 整合 ActiveMQ
前文消息中间件 -- 简介 ActiveMQ 下载.安装 ActiveMQ -- Java 连接 ActiveMQ(点对点) ActiveMQ -- Java 连接 ActiveMQ(发布订阅 To ...
Spring整合Activemq
目录一.Pom.xml 二.Spring配置文件三.队列四.主题一.Pom.xml <dependencies><de ...
Java消息队列-Spring整合ActiveMq
1.概述首先和大家一起回顾一下Java 消息服务,在我之前的博客<Java消息队列-JMS概述>中,我为大家分析了: 消息服务:一个中间件,用于解决两个或多个程序之间的耦合,底层由Jav ...
java 消息队列详解_Java消息队列-Spring整合ActiveMq的详解
本篇文章主要介绍了详解Java消息队列-Spring整合ActiveMq ,小编觉得挺不错的,现在分享给大家,也给大家做个参考.一起跟随小编过来看看吧 1.概述首先和大家一起回顾一下Java 消息服 ...
ActiveMQ学习总结（3）——spring整合ActiveMQ
2019独角兽企业重金招聘Python工程师标准>>> 1.参考文献 Spring集成ActiveMQ配置 Spring JMS异步发收消息 ActiveMQ 2.环境在前面的一篇 ...

spring整合使用activemq

spring整合使用activemq相关推荐

最新文章

热门文章