activemq与spring的整合需要用到线程池。考虑到连接、会话等资源的建立和释放，无须人工操作，全部交给容器来处理。这里通过一个实例讲解activemq与spring如何整合。项目大致是这样的设计：通过jetty构建一个http请求，接收http://localhost:8080/send?msg=xxx的请求，然后将msg作为消息传递给生产者线程，将消息发送到指定的Queue，这里发送消息基本完成，为了验证接收消息，这里配置一个消费者，用来监听实时消息，而不是通过循环来读取消息，因此需要注册一个消息监听器，来监听生产者发送的消息，然后将接收的消息打印出来，完成activemq与spring整合之后发送接收的测试。
1、构建maven项目，并引入依赖库。项目最终项目结构如下：
pom.xml文件如下：
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.xxx.activemq</groupId>
<artifactId>sendmessage</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>sendmessage</name>
<url>http://maven.apache.org</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<spring-version>4.3.4.RELEASE</spring-version>
</properties>
<dependencies>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>test-jetty-servlet</artifactId>
<version>8.1.0.RC5</version>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.21</version>
</dependency>
<dependency>
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
<version>2.6</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
<version>2.4.2</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.15.0</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-beans</artifactId>
<version>${spring-version}</version>
</dependency>
<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>
</dependencies>
</project>
2、主要用到的类介绍
ISendService.java
package com.xxx.activemq.service;
public interface ISendService {
public void send(String msg);
}
SendService.java,将/send?msg=xxx发送过来的msg参数作为消息发送给activemq的一个Queue,这里是brokerQueue,后面的applicationContext.xml配置文件中会配置该Queue。
package com.xxx.activemq.service;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.apache.log4j.Logger;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
public class SendService implements ISendService{
private static final Logger LOG = Logger.getLogger(SendService.class);
private String queueName;
private JmsTemplate jmsTemplate;
private ThreadPoolTaskExecutor threadExecutor;
@Override
public void send(String msg) {
threadExecutor.execute(new Runnable() {
@Override
public void run() {
jmsTemplate.send(queueName,new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(msg);
}
});
LOG.info(String.format("send msg %s to "+queueName+" ok.", msg));
}
});
}
public String getQueueName() {
return queueName;
}
public void setQueueName(String queueName) {
this.queueName = queueName;
}
public JmsTemplate getJmsTemplate() {
return jmsTemplate;
}
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public ThreadPoolTaskExecutor getThreadExecutor() {
return threadExecutor;
}
public void setThreadExecutor(ThreadPoolTaskExecutor threadExecutor) {
this.threadExecutor = threadExecutor;
}
}
package com.xxx.activemq.servlet;
import java.io.IOException;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import com.xxx.activemq.service.SendService;
import com.xxx.activemq.utils.WebContext;
public class SendServlet extends HttpServlet{
/**
*
*/
private static final long serialVersionUID = 1L;
private static SendService sendService;
static{
sendService = (SendService)WebContext.getBean("sendService");
}
@Override
protected void doGet(HttpServletRequest req, HttpServletResponse res)
throws ServletException, IOException {
doPost(req, res);
}
@Override
protected void doPost(HttpServletRequest req, HttpServletResponse res)
throws ServletException, IOException {
String message = req.getParameter("msg");
sendService.send(message);
res.setCharacterEncoding("UTF-8");
res.setContentType("application/json");
res.setStatus(HttpServletResponse.SC_OK);
res.getWriter().println("{\"message\":\"ok\"}");
}
}
package com.xxx.activemq.webapp;
import org.apache.log4j.Logger;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.servlet.ServletContextHandler;
import org.eclipse.jetty.servlet.ServletHolder;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import com.xxx.activemq.servlet.SendServlet;
import com.xxx.activemq.utils.WebContext;
public class App {
private static final int PORT = 8080;
private static final Logger LOG = Logger.getLogger(App.class);
public static void main(String[] args) {
ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");
WebContext.setContext(context);
Server server = new Server(PORT);
ServletContextHandler handler = new ServletContextHandler(ServletContextHandler.SESSIONS);
handler.setContextPath("/");
handler.addServlet(new ServletHolder(new SendServlet()), "/send");
server.setHandler(handler);
try {
LOG.info("webapp initialize success!");
server.start();
server.join();
} catch (Exception e) {
LOG.error("webapp initialize error!");
}
}
}
package com.xxx.activemq.utils;
import org.apache.log4j.Logger;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
public class WebContext implements ApplicationContextAware {
private static final Logger LOG = Logger.getLogger(WebContext.class);
private static ApplicationContext context;
public void setApplicationContext(ApplicationContext context)
throws BeansException {
LOG.info("applicationContext init...");
setContext(context);
}
public static void setContext(ApplicationContext context) {
WebContext.context = context;
}
public static Object getBean(String name){
if(context==null){
LOG.error("context is null");
return null;
}
return context.getBean(name);
}
}package com.xxx.activemq.listener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
import org.apache.log4j.Logger;
public class ConsumerMessageListener implements MessageListener {
private static Logger LOG = Logger.getLogger(ConsumerMessageListener.class);
@Override
public void onMessage(Message msg) {
if(msg instanceof TextMessage){
TextMessage o = (TextMessage)msg;
try {
LOG.info("received message: " + o.getText());
} catch (JMSException e) {
LOG.error("received message for error!");
}
}
}
}配置文件：
applicationContext.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"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.3.xsd">
<bean id="propertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
<property name="locations">
<list>
<value>classpath:config.properties</value>
</list>
</property>
</bean>
<bean id="jmsConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
<property name="connectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="${activemq.brokerURL}"></property>
<property name="useAsyncSend" value="true"/>
<property name="redeliveryPolicy">
<bean class="org.apache.activemq.RedeliveryPolicy">
<property name="initialRedeliveryDelay" value="2000"/>
<property name="useExponentialBackOff" value="true"/>
</bean>
</property>
</bean>
</property>
</bean>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory" ref="jmsConnectionFactory"/>
</bean>
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory"/>
</bean>
<bean id="threadExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<property name="corePoolSize" value="${sendmsgspool.poolsize}" />
<property name="keepAliveSeconds" value="${sendmsgspool.keepaliveseconds}" />
<property name="maxPoolSize" value="${sendmsgspool.maxpoolsize}" />
<property name="rejectedExecutionHandler">
<bean class="java.util.concurrent.ThreadPoolExecutor$DiscardOldestPolicy" />
</property>
</bean>
<bean id="sendService" class="com.xxx.activemq.service.SendService">
<property name="queueName" value="brokerQueue" />
<property name="jmsTemplate" ref="jmsTemplate"/>
<property name="threadExecutor" ref="threadExecutor"/>
</bean>
<!-- 发送消息以上这些配置足够了 -->
<!-- 下面配置收消息的消费者 -->
<bean id="consumerMessageListener" class="com.xxx.activemq.listener.ConsumerMessageListener">
</bean>
<bean id="brokerQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="${activemq.brokerQueue}"/>
</bean>
<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="messageListener" ref="consumerMessageListener"/>
<property name="destination" ref="brokerQueue"/>
</bean>
</beans>
activemq.brokerURL=tcp://localhost:61616
activemq.brokerQueue=brokerQueue
sendmsgspool.keepaliveseconds=300
sendmsgspool.maxpoolsize=50
sendmsgspool.poolsize=15
3、启动App.java，监听8080端口，接收http://localhost:8080/send?msg=xxx的请求，然后将参数msg作为消息利用生产者发送给消息接收者。
项目启动成功日志：
第一次请求：
控制台日志：
第二次请求：
控制台日志：
管理界面查看队列消息：
4、这里重要的就是要理清jmsTemplate,jmsContainer,connectionFactory,jmsConnectionFactory之间的关系。
activemq与spring的整合和java直接编写activemq实例最大的区别在于，connection,session等资源的创建和释放无需人为操作，看不到很直观的API调用。
