spring整合activeMQ

我们先从消息生产者这一端开始整合

创建一个maven工程结构如下



我的pom文件如下

[html] view
plain copy







<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.pcx</groupId>

<artifactId>amqproducer</artifactId>

<version>0.0.1-SNAPSHOT</version>

<packaging>jar</packaging>

<name>amqproducer</name>

<url>http://maven.apache.org</url>

<properties>

<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

<spring.version>3.2.4.RELEASE</spring.version>

</properties>

<dependencies>

<dependency>

<groupId>org.springframework</groupId>

<artifactId>spring-aop</artifactId>

<version>${spring.version}</version>

</dependency>

<dependency>

<groupId>org.springframework</groupId>

<artifactId>spring-web</artifactId>

<version>${spring.version}</version>

</dependency>

<dependency>

<groupId>org.springframework</groupId>

<artifactId>spring-context-support</artifactId>

<version>${spring.version}</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-core</artifactId>

<version>${spring.version}</version>

</dependency>

<dependency>

<groupId>org.springframework</groupId>

<artifactId>spring-jms</artifactId>

<version>${spring.version}</version>

</dependency>

<dependency>

<groupId>org.apache.activemq</groupId>

<artifactId>activemq-all</artifactId>

<version>5.11.1</version>

</dependency>

<dependency>

<groupId>javax.jms</groupId>

<artifactId>jms</artifactId>

<version>1.1</version>

</dependency>

<!-- Test Dependency Begin -->

<dependency>

<groupId>junit</groupId>

<artifactId>junit</artifactId>

<version>4.11</version>

</dependency>

</dependencies>

</project>

我的spring-context.xml文件如下

[html] view
plain copy







<?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="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

<!-- ActiveMQ服务地址 -->

<property name="brokerURL" value="tcp://192.168.1.10:61616" />

</bean>

<!--配置我们要发送的队列名字 -->

<bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">

<constructor-arg value="user.queue"/>

</bean>

<!-- spring发送消息都是使用jmstemplate这个封装过的模板类来操作的-->

<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

<!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->

<property name="connectionFactory" ref="connectionFactory"/>

</bean>

</beans>

接下来我们编写主流程代码

[java] view
plain copy







/**

*

*/

package com.pcx.amqproducer;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MapMessage;

import javax.jms.Message;

import javax.jms.Session;

import org.springframework.context.ApplicationContext;

import org.springframework.context.support.ClassPathXmlApplicationContext;

import org.springframework.jms.core.JmsTemplate;

import org.springframework.jms.core.MessageCreator;

/**

*

* @author scarletbullet

* @version $Id: MySpringProducer.java

*/

public class MySpringProducer {

public static void main(String[] args) {

ApplicationContext context;

context= new ClassPathXmlApplicationContext("classpath:spring/spring-context.xml");

JmsTemplate template = (JmsTemplate) context.getBean("jmsTemplate");

Destination destination = (Destination) context.getBean("destination");

template.send(destination, new MessageCreator() {

public Message createMessage(Session session) throws JMSException {

MapMessage message=session.createMapMessage();

message.setString("userId","12456");

message.setString("userName", "张三");

message.setInt("age", 22);

return message;

}

});

System.out.println("成功发送了一条JMS消息");

}

}

maven clean install 一下后我们运行这个java类



打开控制台我们可以看到消息增加了一条



接下来我们编写一下消费端

同样创建一个maven工程，结构如下



我们在spring-context.xml文件中的配置

[html] view
plain copy







<?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">
<!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供 -->

<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

<!-- ActiveMQ服务地址 -->

<property name="brokerURL" value="tcp://192.168.1.10:61616" />

</bean>

<!-- 接受消息的目的地（配置队列名称） -->

<bean id="destination" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg value="user.queue"/> </bean> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 --> <property name="connectionFactory" ref="connectionFactory"/> </bean> </beans>

接下来我们编写消费者的java代码

[java] view
plain copy







/**

*

*/

package com.pcx.amqconsumer;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MapMessage;

import org.springframework.context.ApplicationContext;

import org.springframework.context.support.ClassPathXmlApplicationContext;

import org.springframework.jms.core.JmsTemplate;

/**

*

* @author scarletbullet

* @version $Id: MySpringConsumer.java

*/

public class MySpringConsumer {

/**

*

* @param args

*/

public static void main(String[] args) {

ApplicationContext context;

context= new ClassPathXmlApplicationContext("classpath:spring-context.xml");

JmsTemplate template = (JmsTemplate) context.getBean("jmsTemplate");

Destination destination = (Destination) context.getBean("destination");

[java] view
plain copy







<span style="white-space:pre"> </span>//同步阻塞等待消息

MapMessage message=(MapMessage)template.receive(destination);

try {

System.out.println(message.getString("userName"));

} catch (JMSException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

}

}

我们运行一下这个类



我们在看一下amq控制台,可以发现刚刚生产的消息被消费掉了



刚刚的消息生产者我们显示的配置了一个destination

[html] view
plain copy







<bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">

<constructor-arg value="user.queue"/>

</bean>

实际上我们可以指定一个默认destinationName在我们的jmsTemplate里面，如下

[html] view
plain copy







<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

<!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->

<property name="connectionFactory" ref="connectionFactory"/>

<property name="defaultDestinationName" value="user.queue"/>

</bean>

我们的消息生产类做出如下修改

[html] view
plain copy







/**

*

*/

package com.pcx.amqproducer;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MapMessage;

import javax.jms.Message;

import javax.jms.Session;

import org.springframework.context.ApplicationContext;

import org.springframework.context.support.ClassPathXmlApplicationContext;

import org.springframework.jms.core.JmsTemplate;

import org.springframework.jms.core.MessageCreator;

/**

*

* @author scarletbullet

* @version $Id: MySpringProducer.java, v 0.1 2016-5-15 下午10:50:38 scarletbullet Exp $

*/

public class MySpringProducer {

public static void main(String[] args) {

ApplicationContext context;

context= new ClassPathXmlApplicationContext("classpath:spring/spring-context.xml");

JmsTemplate template = (JmsTemplate) context.getBean("jmsTemplate");

//Destination destination = (Destination) context.getBean("destination");

template.send( new MessageCreator() {

public Message createMessage(Session session) throws JMSException {

MapMessage message=session.createMapMessage();

message.setString("userId","12456");

message.setString("userName", "张三");

message.setInt("age", 22);

return message;

}

});

System.out.println("成功发送了一条JMS消息");

}

}