基于Maven的ActiveMQ+spring 多消息队列配置实例

一、前言：

本文主要是示范基于Maven的ActiveMQ+spring的多消息队列简单使用，基于ActiveMQ消息代理的Spring JMS消息配置，以springmvc的使用消息队列

二、环境：

在apache官方网站下载后，解压缩安装包，点击apache-activemq-5.9.0\bin\activemq.bat运行即可。运行后浏览器输入http://localhost:8161/表明安装成功， 这时就可以使用它进行消息代理了。

三、配置部分

1. maven配置。

引入依赖jar包。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/maven-v4_0_0.xsd"> 
<modelVersion>4.0.0</modelVersion>
<groupId>com.worldsconnection</groupId>
<artifactId>activeMQDemao</artifactId>
<packaging>war</packaging>
<version>0.0.1-SNAPSHOT</version>
<name>gzframe Maven Webapp</name>
<url>http://maven.apache.org</url>

<!-- 版本管理 -->
<properties>
<springframework>4.0.2.RELEASE</springframework>
<javax.servlet>3.1.0</javax.servlet>
</properties>

<dependencies>

<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.10</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>jstl</groupId>
<artifactId>jstl</artifactId>
<version>1.2</version>
</dependency>
<dependency>
<groupId>javax.servlet</groupId>
<artifactId>servlet-api</artifactId>
<version>${javax.servlet}</version>
</dependency>

<!-- spring -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
<version>${springframework}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>${springframework}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-tx</artifactId>
<version>${springframework}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-webmvc</artifactId>
<version>${springframework}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>${springframework}</version>
</dependency>
<!-- xbean 如<amq:connectionFactory /> -->
<dependency>
<groupId>org.apache.xbean</groupId>
<artifactId>xbean-spring</artifactId>
<version>3.16</version>
</dependency>

<!-- activemq -->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.9.0</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
<version>5.9.0</version>
</dependency>

</dependencies>
<build>
<finalName>activeMQDemao</finalName>
</build>
</project>

2. spring配置（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:amq="http://activemq.apache.org/schema/core"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.8.0.xsd">
<!-- Activemq connection factory -->
<bean id="amqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<constructor-arg index="0" value="tcp://192.168.0.100:61616" />
<property name="useAsyncSend" value="true" />
</bean>

<!-- ConnectionFactory Definition -->
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<constructor-arg ref="amqConnectionFactory" />
<property name="sessionCacheSize" value="100" />
</bean>

<!-- Default Destination Queue Definition -->
<bean id="toMSQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg index="0" value="toMSQueue" />
</bean>
<bean id="toLDYQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg index="0" value="toLDYQueue" />
</bean>

<!-- JmsTemplate Definition -->
<bean id="jmsTemplateToMS" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory" />
<property name="defaultDestination" ref="toMSQueue" />
<property name="receiveTimeout" value="10000" />
</bean>
<bean id="jmsTemplateToLDY" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory" />
<property name="defaultDestination" ref="toLDYQueue" />
<property name="receiveTimeout" value="10000" />
</bean>

<!-- 配置消息队列监听者（Queue） -->
<bean id="queueMessageListenerToMS" class="com.wanle.ms.mq.QueueMessageListenerToMS" />
<bean id="queueMessageListenerToLDY" class="com.wanle.ms.mq.QueueMessageListenerToLDY" />

<!-- 显示注入消息监听容器（Queue），配置连接工厂，监听的目标是demoQueueDestination，监听器是上面定义的监听器 -->
<bean id="queueListenerContainerToMS"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="toMSQueue" />
<property name="messageListener" ref="queueMessageListenerToMS" />
</bean>
<bean id="queueListenerContainerToLDY"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="toLDYQueue" />
<property name="messageListener" ref="queueMessageListenerToLDY" />
</bean>
</beans>

3. 加载springContext.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:context="http://www.springframework.org/schema/context"
xmlns:mvc="http://www.springframework.org/schema/mvc"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/s
4000
chema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd"> 
<context:component-scan base-package="com.worldsconnection.mq" />
<mvc:annotation-driven />

</beans>

4. web.xml中加载spring-activemq.xml 和springContext.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:context="http://www.springframework.org/schema/context"
xmlns:mvc="http://www.springframework.org/schema/mvc"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd">
<context:component-scan base-package="com.worldsconnection.mq" />
<mvc:annotation-driven />

</beans>

四：代码部分：

1.生产者服务（ProducerService.java）

package com.wanle.ms.mq;

import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;

@Service
public class ProducerService {

@Resource(name = "jmsTemplateToMS")
private JmsTemplate jmsTemplateToMS;

@Resource(name = "jmsTemplateToLDY")
private JmsTemplate jmsTemplateToLDY;

/**
* 向指定队列发送消息
*/
public void sendMessage(Destination destination, final String msg) {
System.out.println("向队列" + destination.toString() + "发送了消息------------" + msg);
jmsTemplateToMS.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(msg);
}
});
}

/**
* 向默认队列发送消息
*/
public void sendMessage(final String msg) {
String destination = jmsTemplateToMS.getDefaultDestination().toString();
System.out.println("向队列" + destination + "发送了消息------------" + msg);
jmsTemplateToMS.send(new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(msg);
}
});
}

}

2.消费者服务（ConsumerService.java）

package com.wanle.ms.mq;

import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.TextMessage;

import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;

@Service
public class ConsumerService {

@Resource(name = "jmsTemplateToMS")
private JmsTemplate jmsTemplateToMS;

@Resource(name = "jmsTemplateToLDY")
private JmsTemplate jmsTemplateToLDY;

/**
* 接收消息
*/
public TextMessage receive(Destination destination) {
TextMessage tm = (TextMessage) jmsTemplateToMS.receive(destination);
try {
System.out.println("从队列" + destination.toString() + "收到了消息：\t" + tm.getText());
} catch (JMSException e) {
e.printStackTrace();
}
return tm;
}

}

3. 消费者监听器

一般mq的消费者，都是自动消费掉生产者的业务请求的。 所以一般都配置为listener。 只有有生产者生产出业务，消费者都立即消费掉。 

业务消费者1（QueueMessageListenerToMS.java）

</pre><p><pre name="code" class="java">package com.wanle.ms.mq;

import javax.annotation.Resource;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

public class QueueMessageListenerToMS implements MessageListener {

@Resource(name = "jmsTemplateToMS")
private JmsTemplate jmsTemplateToMS;

@Resource(name = "jmsTemplateToLDY")
private JmsTemplate jmsTemplateToLDY;

//当收到消息后，自动调用该方法
@Override
public void onMessage(Message message) {

TextMessage tm = (TextMessage) message;
try {
System.out.println("(MS::)QueueMessageListener监听到了文本消息：\t" + tm.getText());
//这里有个业务场景，QueueMessageListenerToMS 处理消息队列时候，会再次发送其他业务消息。
String destination = jmsTemplateToLDY.getDefaultDestination().toString();
System.out.println("向队列" + destination + "发送了消息------------" + tm.getText());
jmsTemplateToLDY.send(new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(tm.getText());
}
});
//do something ...
} catch (JMSException e) {
e.printStackTrace();
}
}

}

业务消费者2 （QueueMessageListenerToLDY.java）

package com.wanle.ms.mq;

import javax.annotation.Resource;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

public class QueueMessageListenerToLDY implements MessageListener {

@Resource(name = "jmsTemplateToMS")
private JmsTemplate jmsTemplateToMS;

@Resource(name = "jmsTemplateToLDY")
private JmsTemplate jmsTemplateToLDY;

//当收到消息后，自动调用该方法
@Override
public void onMessage(Message message) {

TextMessage tm = (TextMessage) message;
try {
System.out.println("(LDY::)QueueMessageListener监听到了文本消息：\t" + tm.getText());
//do something ...
} catch (JMSException e) {
e.printStackTrace();
}
}

}

4.服务调用者（TestController.java）

package com.wanle.ms.controller.login;

import java.text.SimpleDateFormat;
import java.util.Date;

import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.TextMessage;

import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;

import com.wanle.ms.mq.ConsumerService;
import com.wanle.ms.mq.ProducerService;

@Controller
@RequestMapping("test")
public class TestController {

//队列名
@Resource(name="toMSQueue")
private Destination toMSQueue;
@Resource(name="toLDYQueue")
private Destination toLDYQueue;

//队列消息生产者
@Resource(name="producerService")
private ProducerService producer;

//队列消息消费者
@Resource(name="consumerService")
private ConsumerService consumer;

@RequestMapping(value="/producer")
public String producer(){
System.out.println("------------go producer");
Date now = new Date();
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String time = dateFormat.format( now );
System.out.println(time);
producer.sendMessage(toMSQueue, time);
return null;
}

@RequestMapping(value="/receive")
public String queue_receive() throws JMSException {
System.out.println("------------receive message");

TextMessage tm = consumer.receive(toMSQueue);
String message = tm.getText();
System.out.println(message);
return null;
}

}

五：测试：

部署应用，访问localhost:8080/test/producer 即可看到效果。

以上内容为实际项目内容，绝对可靠