码迷，mamicode.com
首页 > 编程语言 > 详细

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

时间：2016-04-10 14:39:20      阅读：265      评论：0      收藏：0      [点我收藏+]

标签：

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的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.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.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.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项目案例

标签：

原文地址：http://blog.csdn.net/tanggao1314/article/details/51107382
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！