JMS(java消息服务)整合Spring项目案例

转载自云栖社区

摘要： Sprng-jms消息服务小项目所需的包: spring的基础包 spring-jms-xx包 spring-message–xx包 commons-collection-xx包 commons-pool2-xx包 aop切面的包: spring-aop,spring-aspect...

Sprng-jms消息服务小项目

所需的包:
spring的基础包
spring-jms-xx包
spring-message–xx包
commons-collection-xx包
commons-pool2-xx包
aop切面的包: spring-aop,spring-aspect,aopalliance,aspectjrt.jar,aspectjweaver.jar

配置:
1.配置ConnectionFactory
2.配置jmsTemplate;
3.配置Destination

4.配置listener
5.配置container

话不多说，直接上代码
前提是你已经开启了activemq服务
看一下项目架构

JMS(java消息服务)整合Spring项目案例

临听器有两种实现方案: 一种是采用原生的jms的MessageListener
另一种是采用spring的方案:SessionAwareMessageListener
前一种只能消费消息,不能发送回送消息
后一种可以再接到消息后，给生产者回送消息,它自己即是生产者也是消费者
所有所需的连接工、目的地等都是通过Spring注入的,具体看后面的配置文件

发送消息的接口：

package com.tg.service;

public interface PersonService {

    public void sendMessage(String message);

}

生产者一：

package com.tg.service;

import javax.annotation.Resource;

import javax.jms.DeliveryMode;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.Session;

import org.springframework.context.annotation.Scope;

import org.springframework.jms.core.JmsTemplate;

import org.springframework.jms.core.MessageCreator;

import org.springframework.stereotype.Service;

    @Service("personServiceImpl")

    @Scope("prototype")

    public class PersonServiceImpl implements PersonService {

        private Destination destination;

        private JmsTemplate jsmTemplate;

        private Destination replyQueueDestination;

        @Override

        public void sendMessage(final String message) {

            System.out.println("生产者发送消息"+ message);

            //回调

            jsmTemplate.send(destination, new MessageCreator(){

                @Override

                public Message createMessage(Session session) throws JMSException {

                    //创建一个文本消息

                    Message msg = session.createTextMessage(message);

                    //指定为非持久化方式

                    msg.setJMSDeliveryMode(   DeliveryMode.NON_PERSISTENT   );

                    return msg;

                }

            });

        }

        @Resource(name="queueDestination")

        public void setDestination(Destination destination) {

            this.destination = destination;

        }

        @Resource(name="jmsTemplate")

        public void setJsmTemplate(JmsTemplate jsmTemplate) {

            this.jsmTemplate = jsmTemplate;

        }

        @Resource(name = "replyQueueDestination")

        public void setReplyDestination(Destination replyDestination) {

            this.replyQueueDestination = replyDestination;

        }

}

消费者一：

package com.tg.service;

import javax.jms.Message;

import javax.jms.MessageListener;

import javax.jms.TextMessage;

//临听器有两种实现方案: 一种是采用原生的jms的MessageListener

//                    另一种是采用spring的方案:SessionAwareMessageListener 

//注意: 这里的MessageListener接口是  jms的接口

public class ConsumerMessageListener1 implements MessageListener {

    @Override

    public void onMessage(Message message) {

        if(  message instanceof TextMessage){

            TextMessage text=(TextMessage) message;

            System.out.println("接收到的消息是一个文本消息:"+ text);

            //这种方式无法回复   所以采用第二种

        }

    }

}

生产者二：

package com.tg.service;

import javax.annotation.Resource;

import javax.jms.DeliveryMode;

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.springframework.jms.core.JmsTemplate;

import org.springframework.jms.core.MessageCreator;

import org.springframework.stereotype.Service;

@Service("personServiceImpl2")

public class PersonServiceImpl2 implements PersonService {

    private Destination destination;   //用于存发送信息的队列

    private JmsTemplate jsmTemplate;   //jms操作模板

    private Destination replyQueueDestination;   //用于存回复信息的队列,

    @Override

    public void sendMessage(final String message) {

        System.out.println("生产者2->发送消息" + message);

        // 回调

        jsmTemplate.send(destination, new MessageCreator() {

            @Override

            public Message createMessage(Session session) throws JMSException {

                Message msg = session.createTextMessage(message);

                // 设置回复的信息的目的地.

                msg.setJMSReplyTo(replyQueueDestination);

                // 设置发送的信息类型 为非持久化信息

                msg.setJMSDeliveryMode(DeliveryMode.NON_PERSISTENT);

                //创建一个消费者，用于接收对方回复的信息   注意，这个消费者临听   replyDestination

                MessageConsumer comsumer2 = session.createConsumer(replyQueueDestination);

                comsumer2.setMessageListener(new MessageListener() {

                    public void onMessage(Message m) {

                        try {

                            System.out.println("接收到的回复信息:" + ((TextMessage) m).getText());

                        } catch (JMSException e) {

                            e.printStackTrace();

                        }

                    }

                });

                return msg;

            }

        });

    }

    @Resource(name = "replyQueueDestination")

    public void setReplyDestination(Destination replyDestination) {

        this.replyQueueDestination = replyDestination;

    }

    @Resource(name = "sendQueueDestination")

    public void setDestination(Destination destination) {

        this.destination = destination;

    }

    @Resource(name = "jmsTemplate")

    public void setJsmTemplate(JmsTemplate jsmTemplate) {

        this.jsmTemplate = jsmTemplate;

    }

}

消费者二：

package com.tg.service;

import javax.annotation.Resource;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageProducer;

import javax.jms.Session;

import javax.jms.TextMessage;

import org.springframework.jms.listener.SessionAwareMessageListener;

import org.springframework.stereotype.Component;

/**

 * SessionAwareMessageListener：是由spring提供，它可以在回调方法中传入session，以此回送信息到生产者

 * @author Administrator

 *

 */

@Component("consumerMessageListener2")

public class ConsumerMessageListener2 implements SessionAwareMessageListener<TextMessage> {

    private Destination destination;

    @Override

    public void onMessage(TextMessage message, Session session) throws JMSException {

            System.out.println("接收到的消息是一个文本消息:"+ message.getText());

            //通过session 创建  producer对象，再回送信息

            //从message中取出信息回送的目的地,以便创建生产者.

            MessageProducer producer=session.createProducer( message.getJMSReplyTo()   );

            //创建一条消息

            Message textMessage=session.createTextMessage(  "生产者发过来的信息已经处理完毕，game over..."   );

            //调用发送

            producer.send(textMessage);

    }

    @Resource(name="sendQueueDestination")

    public void setDestination(Destination destination) {

        this.destination = destination;

    }

}

Spring的配置文件：

<?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:aop="http://www.springframework.org/schema/aop"

    xmlns:context="http://www.springframework.org/schema/context" xmlns:tx="http://www.springframework.org/schema/tx"

    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.xsd

        http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd

        http://www.springframework.org/schema/tx

        http://www.springframework.org/schema/tx/spring-tx.xsd">

    <!-- 启用注解解析器 -->

    <context:annotation-config />

    <!-- 因为采用了混合解析方式( 有一部分配置在xml中,有一部分在java类中，所以要让spring的注解解析器去扫描包 -->

    <context:component-scan base-package="com.*" />

    <!-- 启用aspectj的注解方式的代理 -->

    <aop:aspectj-autoproxy />

    <!-- 创建一个真正的基于 jsm提供者的联接工厂 -->

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

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

    </bean>

    <!-- ActiveMQ联接池的方案 -->

    <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">

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

        <property name="maxConnections" value="100" />

    </bean>

    <!-- 创建spring联接工厂 -->

    <bean id="singleConnectionFactory"

        class="org.springframework.jms.connection.SingleConnectionFactory">

        <property name="targetConnectionFactory" ref="pooledConnectionFactory" />

    </bean>

    <!-- <bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"

        <property name="targetConnectionFactory" ref="pooledConnectionFactory" />

        </bean> -->

    <!-- 配置jmsTemplate -->

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

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

    </bean>

    <!-- 配置目的地: 这有两种：一种是 Queue对应是 P2P模式,另一种是 Topic 对应的是 发布/订阅模式, -->

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

        <constructor-arg>

            <value>myqueue</value>

        </constructor-arg>

    </bean>

    <!-- <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">

        <constructor-arg> <value>mytopic</value> </constructor-arg> </bean> -->

    <!-- 配置临听器 -->

    <bean id="consumerMessageListener1" class="com.tg.service.ConsumerMessageListener1">

    </bean>

    <!-- 配置临听器运行时所在的容器  让Listener运行在这个容器,这样

    只要有消息过来,就会回调

    -->

    <bean id="jmsContainer"

        class="org.springframework.jms.listener.DefaultMessageListenerContainer">

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

        <property name="destination" ref="queueDestination" />

        <property name="messageListener" ref="consumerMessageListener1" />

    </bean>

    <!-- 以下是第二种方案 -->

    <!-- 以下用于存放生产者发送的信息 -->

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

        <constructor-arg>

            <value>sendQueue1</value>

        </constructor-arg>

    </bean>

    <!--  以下用于存放消费者回复的信息  -->

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

        <constructor-arg>

            <value>replyQueue1</value>

        </constructor-arg>

    </bean>

    <!-- 配置   sessionAware的临听器

    <bean id="consumerMessageListener2" class="com.yc.jms5.ConsumerMessageListener2" >

        <property name="destination" ref="sendQueueDestination" />

        <property name="replyDestination" ref="replyQueueDestination" />

    </bean>

     -->

    <!-- 配置  consumerMessageListener2的容器 -->

    <bean id="sessionAwareListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer" >

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

            <property name="destination" ref="sendQueueDestination" />

            <property name="messageListener" ref="consumerMessageListener2" />

    </bean>

</beans>

第一种的测试类：

package com.tg.test;

import org.apache.xbean.spring.context.ClassPathXmlApplicationContext;

import org.springframework.context.ApplicationContext;

import com.tg.service.PersonService;

//测试生产者发送了一个消息，消费者接收

//整合spring    http://haohaoxuexi.iteye.com/blog/1893038

public class Test1 {

    public static void main(String[] args) {

        ApplicationContext ac=new ClassPathXmlApplicationContext("applicationContext.xml");

        PersonService ps=(PersonService) ac.getBean("personServiceImpl");

        for( int i=0;i<10;i++){

            Thread t=new Thread(new User(ps,"汤"+i));

            t.start();

        }

    }

}

class User implements Runnable{

    private PersonService ps;

    private String name;

    public User(PersonService ps, String name) {

        super();

        this.ps = ps;

        this.name = name;

    }

    @Override

    public void run() {

        for( int i=0;i<10;i++){

            ps.sendMessage(name+"向你问好");

        }

    }

}

运行结果：

第二种方案测试类：

package com.tg.test;

import org.apache.xbean.spring.context.ClassPathXmlApplicationContext;

import org.springframework.context.ApplicationContext;

import com.tg.service.PersonService;

//测试生产者发送了一个消息，消费者接收后，再回复一个信息到生产者，生产者接收到后，显示这个回复的信息

public class Test2 {

    public static void main(String[] args) {

        ApplicationContext ac=new ClassPathXmlApplicationContext("applicationContext.xml");

        PersonService ps=(PersonService) ac.getBean("personServiceImpl2");

        for( int i=0;i<10;i++){

            Thread t=new Thread(new User(ps,":   汤1"+i));

            t.start();

        }

    }

}

运行结果：
JMS(java消息服务)整合Spring项目案例