ActiveMQ之整合spring
Posted 小~虎
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了ActiveMQ之整合spring相关的知识，希望对你有一定的参考价值。
ActiveMQ可以轻松的与Spring进行整合，Spring提供了一系列的接口类，非常的好用！
比如异步消息数据、异步发送邮件、异步消息查询等
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.11.1</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
<version>5.11.1</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>${spring.version}</version>
</dependency>
先引入两个activeMQ的jar包，
然后进行spring-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">
<!-- 第三方MQ工厂: ConnectionFactory -->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<!-- ActiveMQ Address -->
<property name="brokerURL" value="${activemq.brokerURL}" />
<property name="userName" value="${activemq.userName}"></property>
<property name="password" value="${activemq.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="${activemq.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="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<property name="connectionFactory" ref="connectionFactory"/>
<property name="defaultDestinationName" value="${activemq.queueName}"></property>
</bean>
</beans>
创建好第三方MQ工厂: ConnectionFactory
然后创建生产者MQProducer ：
package bhz.mq;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;
import bhz.entity.Mail;
import com.alibaba.fastjson.JSONObject;
/**
* <B>系统名称：</B><BR>
* <B>模块名称：</B><BR>
* <B>中文类名：</B><BR>
* <B>概要说明：</B><BR>
* @author jxh
*/
@Service("mqProducer")
public class MQProducer {
private JmsTemplate jmsTemplate;
public JmsTemplate getJmsTemplate() {
return jmsTemplate;
}
@Autowired
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
/**
* <B>方法名称：</B>发送邮件信息对象<BR>
* <B>概要说明：</B>发送邮件信息对象<BR>
* @param mail
*/
public void sendMessage(final Mail mail) {
jmsTemplate.send(new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(JSONObject.toJSONString(mail));
}
});
}
}
看一下mail实体类：
package bhz.entity;
public class Mail {
/** 发件人 **/
private String from;
/** 收件人 **/
private String to;
/** 主题 **/
private String subject;
/** 邮件内容 **/
private String content;
public Mail(){}
public Mail(String from, String to, String subject, String content) {
super();
this.from = from;
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;
}
}
下面进行消息的发送，通过junit进行发送测试：
package bhz.test;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import bhz.entity.Mail;
import bhz.mq.MQProducer;
/**
* <B>系统名称：</B><BR>
* <B>模块名称：</B><BR>
* <B>中文类名：</B><BR>
* <B>概要说明：</B><BR>
* @author jxh
*/
@ContextConfiguration(locations = {"classpath:spring-context.xml" })
@RunWith(SpringJUnit4ClassRunner.class)
public class TestProducer {
@Autowired
private MQProducer mqProducer;
@Test
public void send(){
Mail mail = new Mail();
mail.setTo("[email protected]");
mail.setSubject("异步发送邮件");
mail.setContent("Hi,This is a message!");
this.mqProducer.sendMessage(mail);
System.out.println("发送成功..");
}
}
下面看一下消费者的这个项目的配置activemq-consumer：
这个项目多一个config
## ActiveMQ Configuration activemq.brokerURL=tcp\://192.168.1.200\:61616 activemq.userName=bhz activemq.password=bhz activemq.pool.maxConnections=10 #queueName activemq.queueName=mailqueue ## SMTP Configuration mail.host=smtp.163.com ##mail.port=21 mail.username=***@163.com mail.password= mail.smtp.auth=true mail.smtp.timeout=30000 mail.default.from=***@163.com
看一下消费者的consumer的spring-activemq.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">
<!-- 第三方MQ工厂: ConnectionFactory -->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<!-- ActiveMQ服务地址 -->
<property name="brokerURL" value="${activemq.brokerURL}" />
<property name="userName" value="${activemq.userName}"></property>
<property name="password" value="${activemq.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="${activemq.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="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<property name="connectionFactory" ref="connectionFactory"/>
<property name="defaultDestinationName" value="${activemq.queueName}"></property>
</bean>
<!--这个是目的地:mailQueue -->
<bean id="mailQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg>
<value>${activemq.queueName}</value>
</constructor-arg>
</bean>
<!-- 配置自定义监听：MessageListener -->
<bean id="mailQueueMessageListener" class="bhz.mq.MailQueueMessageListener"></bean>
<!-- 将连接工厂、目标对了、自定义监听注入jms模板 -->
<bean id="sessionAwareListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="mailQueue" />
<property name="messageListener" ref="mailQueueMessageListener" />
</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: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"
xmlns:cache="http://www.springframework.org/schema/cache"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.2.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-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/aop http://www.springframework.org/schema/aop/spring-aop-3.2.xsd
http://www.springframework.org/schema/cache http://www.springframework.org/schema/cache/spring-cache-3.2.xsd">
<!-- Spring提供的发送电子邮件的高级抽象类 -->
<bean id="mailSender" class="org.springframework.mail.javamail.JavaMailSenderImpl">
<property name="host" value="${mail.host}" />
<property name="username" value="${mail.username}" />
<property name="password" value="${mail.password}" />
<property name="defaultEncoding" value="UTF-8"></property>
<property name="javaMailProperties">
<props>
<prop key="mail.smtp.auth">${mail.smtp.auth}</prop>
<prop key="mail.smtp.timeout">${mail.smtp.timeout}</prop>
</props>
</property>
</bean>
<bean id="simpleMailMessage" class="org.springframework.mail.SimpleMailMessage">
<property name="from">
<value>${mail.default.from}</value>
</property>
</bean>
<!-- 配置线程池 -->
<bean id="threadPool" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<!-- 线程池维护线程的最少数量 -->
<property name="corePoolSize" value="5" />
<!-- 线程池维护线程所允许的空闲时间 -->
<property name="keepAliveSeconds" value="30000" />
<!-- 线程池维护线程的最大数量 -->
<property name="maxPoolSize" value="50" />
<!-- 线程池所使用的缓冲队列 -->
<property name="queueCapacity" value="100" />
</bean>
</beans>
mail实体类和前面一样，
看一下接收消息的Listener，
package bhz.mq;
import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.listener.SessionAwareMessageListener;
import org.springframework.stereotype.Component;
import bhz.entity.Mail;
import bhz.service.MailService;
import com.alibaba.fastjson.JSONObject;
/**
* <B>系统名称：</B><BR>
* <B>模块名称：</B><BR>
* <B>中文类名：</B><BR>
* <B>概要说明：</B><BR>
* @author jxh
*/
@Component
public class MailQueueMessageListener implements SessionAwareMessageListener<Message> {
@Autowired
private JmsTemplate jmsTemplate;
@Autowired
private Destination mailQueue;
@Autowired
private MailService mailService;
public synchronized void onMessage(Message message, Session session) {
try {
TextMessage msg = (TextMessage) message;
final String ms = msg.getText();
System.out.println("收到消息：" + ms);
//转换成相应的对象
Mail mail = JSONObject.parseObject(ms, Mail.class);
if (mail == null) {
return;
}
try {
//执行发送业务
mailService.mailSend(mail);
} catch (Exception e) {
// 发送异常，重新放回队列
// jmsTemplate.send(mailQueue, new MessageCreator() {
// @Override
// public Message createMessage(Session session) throws JMSException {
// return session.createTextMessage(ms);
// }
// });
e.printStackTrace();
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
最后调用发送邮件的service，进行邮件发送：
package bhz.service;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.mail.MailException;
import org.springframework.mail.SimpleMailMessage;
import org.springframework.mail.javamail.JavaMailSender;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import bhz.entity.Mail;
@Service("mailService")
public class MailService {
@Autowired
private JavaMailSender mailSender;
@Autowired
private SimpleMailMessage simpleMailMessage;
@Autowired
private ThreadPoolTaskExecutor threadPool;
/**
* <B>方法名称：</B>发送邮件<BR>
* <B>概要说明：</B>发送邮件<BR>
* @param mail
*/
public void mailSend(final Mail mail) {
threadPool.execute(new Runnable() {
public void run() {
try {
simpleMailMessage.setFrom(simpleMailMessage.getFrom());
simpleMailMessage.setTo(mail.getTo());
simpleMailMessage.setSubject(mail.getSubject());
simpleMailMessage.setText(mail.getContent());
mailSender.send(simpleMailMessage);
} catch (MailException e) {
e.printStackTrace();
throw e;
}
}
});
}
}
邮件配置是指完成后，就可以正常发送邮件了，这就是通过activemq异步发送邮件。
以上是关于ActiveMQ之整合spring的主要内容，如果未能解决你的问题，请参考以下文章