前面介绍了ActiveMQ的基本安装使用，并写了简单的生产者、消费者。下面主要介绍ActiveMQ的消费者的Listener、Spring整合ActiveMQ。
一、消费者Listener
之前创建的消费者，接收消息的时候都是直接使用consumer.receive()，每次消费一条数据，启动一次获取一次，十分的木讷。实际开发工作中，基本不会使用此种方式，一般，消息的消费者都是持续监听目标队列Queue或者主题Topic，主要应用程序不主动关闭，会一直监听消费消息数据。
JMS listener的消费者：
package com.cfang.mq.simpleCase;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class ConsumerListener {
public static void main(String[] args) {
ConsumerListener consumerListener = new ConsumerListener();
consumerListener.listenMessage();
}
public void listenMessage() {
ConnectionFactory factory = null;
Connection connection = null;
Session session = null;
Destination destination = null;
MessageConsumer consumer = null;
try {
factory = new ActiveMQConnectionFactory("admin", "admin", "tcp://172.31.31.160:61616");
connection = factory.createConnection();
connection.start();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
destination = session.createQueue("tp_simple_queue");
consumer = session.createConsumer(destination);
consumer.setMessageListener(new MessageListener() {
public void onMessage(Message message) {
try {
TextMessage textMessage = (TextMessage) message;
System.out.println(textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
//阻塞代码，模拟应用程序不关闭。如果关闭了，那监听也自动停止了
System.in.read();
} catch (Exception e) {
e.printStackTrace();
} finally {
if(consumer != null){
try {
consumer.close();
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
if(session != null){
try {
session.close();
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
if(connection != null){
try {
connection.close();
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
}
二、Spring整合
Spring整合ActiveMQ非常的便捷，Spring提供了JmsTemplate对其进行操作，非常的方便，下面从配置文件到程序代码逐步介绍。
1、Spring-jms配置文件
<?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:tx="http://www.springframework.org/schema/tx"
xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:amq="http://activemq.apache.org/schema/core"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.1.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.1.xsd
http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
<context:component-scan base-package="com.cfang.amq">
</context:component-scan>
<!-- 配置ActiveMQConnectionFactory连接工厂对象 -->
<amq:connectionFactory id="amqConnectionFactory" brokerURL="tcp://172.31.31.160:61616" userName="admin" password="admin"/>
<!-- 配置connectionFactory的连接池信息 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="amqConnectionFactory"></property>
<property name="maxConnections" value="10"></property>
</bean>
<!-- 带有缓存功能的连接工厂，Session缓存大小可配置 -->
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory" ref="pooledConnectionFactory"></property>
<property name="sessionCacheSize" value="100"></property>
</bean>
<!-- 配置JmsTemplate -->
<bean id="template" class="org.springframework.jms.core.JmsTemplate">
<!-- 给定连接工厂, 必须是spring创建的连接工厂. -->
<property name="connectionFactory" ref="connectionFactory"></property>
<!-- 可选 - 默认目的地命名 -->
<property name="defaultDestinationName" value="tp_simple_queue"></property>
</bean>
<!-- 配置生产者Producer -->
<bean id="springProducer" class="com.cfang.amq.SpringProducer"/>
<!-- 配置消费listener -->
<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"></property>
<property name="destinationName" value="tp_simple_queue"/>
<property name="messageListener" ref="springConsumer"></property>
<property name="concurrentConsumers" value="1"/>
</bean>
<!-- 消费者 -->
<bean id="springConsumer" class="com.cfang.amq.SpringConsumer"/>
</beans>
2、单元测试
package com.cfang.prebo.activemq;
import java.io.IOException;
import java.util.Scanner;
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 com.cfang.amq.SpringProducer;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = {"classpath:applicationContext-jms.xml" })
public class SpringListenerTest {
@Autowired
private SpringProducer springConsumer;
@Test
public void start() throws Exception {
System.out.println("======start");
//发送消息
Scanner scanner = new Scanner(System.in);
while(true) {
System.out.print("producer send msg : ");
String line = scanner.nextLine();
if("exit".equals(line)) {
break;
}
springConsumer.sendMsg(null, line);
}
//阻塞
// System.in.read();
}
}
3、运行结果：
