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

简介:

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();
        }
    }

}

运行结果
这里写图片描述

相关实践学习
消息队列+Serverless+Tablestore:实现高弹性的电商订单系统
基于消息队列以及函数计算,快速部署一个高弹性的商品订单系统,能够应对抢购场景下的高并发情况。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
2月前
|
监控 Java 应用服务中间件
高级java面试---spring.factories文件的解析源码API机制
【11月更文挑战第20天】Spring Boot是一个用于快速构建基于Spring框架的应用程序的开源框架。它通过自动配置、起步依赖和内嵌服务器等特性,极大地简化了Spring应用的开发和部署过程。本文将深入探讨Spring Boot的背景历史、业务场景、功能点以及底层原理,并通过Java代码手写模拟Spring Boot的启动过程,特别是spring.factories文件的解析源码API机制。
108 2
|
11天前
|
存储 监控 Java
JAVA线程池有哪些队列? 以及它们的适用场景案例
不同的线程池队列有着各自的特点和适用场景,在实际使用线程池时,需要根据具体的业务需求、系统资源状况以及对任务执行顺序、响应时间等方面的要求,合理选择相应的队列来构建线程池,以实现高效的任务处理。
88 12
|
14天前
|
Java Spring
Java Spring Boot监听事件和处理事件
通过上述步骤,我们可以在Java Spring Boot应用中实现事件的发布和监听。事件驱动模型可以帮助我们实现组件间的松耦合,提升系统的可维护性和可扩展性。无论是处理业务逻辑还是系统事件,Spring Boot的事件机制都提供了强大的支持和灵活性。希望本文能为您的开发工作提供实用的指导和帮助。
67 15
|
12天前
|
监控 JavaScript 数据可视化
建筑施工一体化信息管理平台源码,支持微服务架构,采用Java、Spring Cloud、Vue等技术开发。
智慧工地云平台是专为建筑施工领域打造的一体化信息管理平台,利用大数据、云计算、物联网等技术,实现施工区域各系统数据汇总与可视化管理。平台涵盖人员、设备、物料、环境等关键因素的实时监控与数据分析,提供远程指挥、决策支持等功能,提升工作效率,促进产业信息化发展。系统由PC端、APP移动端及项目、监管、数据屏三大平台组成,支持微服务架构,采用Java、Spring Cloud、Vue等技术开发。
|
1月前
|
Java 开发者 微服务
Spring Boot 入门:简化 Java Web 开发的强大工具
Spring Boot 是一个开源的 Java 基础框架,用于创建独立、生产级别的基于Spring框架的应用程序。它旨在简化Spring应用的初始搭建以及开发过程。
75 6
Spring Boot 入门:简化 Java Web 开发的强大工具
|
2月前
|
人工智能 前端开发 Java
基于开源框架Spring AI Alibaba快速构建Java应用
本文旨在帮助开发者快速掌握并应用 Spring AI Alibaba,提升基于 Java 的大模型应用开发效率和安全性。
269 12
基于开源框架Spring AI Alibaba快速构建Java应用
|
2月前
|
jenkins Java 测试技术
如何使用 Jenkins 自动发布 Java 代码,通过一个电商公司后端服务的实际案例详细说明
本文介绍了如何使用 Jenkins 自动发布 Java 代码,通过一个电商公司后端服务的实际案例,详细说明了从 Jenkins 安装配置到自动构建、测试和部署的全流程。文中还提供了一个 Jenkinsfile 示例,并分享了实践经验,强调了版本控制、自动化测试等关键点的重要性。
96 3
|
2月前
|
Java Maven Spring
Spring 小案例体验创建对象的快感
本文介绍了如何在IDEA中创建一个Spring项目,包括项目创建、配置pom.xml文件以引入必要的依赖、编写实体类HelloSpring及其配置文件applicationContext.xml,最后通过测试类TestHelloSpring展示如何使用Spring的bean创建对象并调用方法。
40 0
|
2月前
|
Java 数据库连接 API
Spring 框架的介绍(Java EE 学习笔记02)
Spring是一个由Rod Johnson开发的轻量级Java SE/EE一站式开源框架,旨在解决Java EE应用中的多种问题。它采用非侵入式设计,通过IoC和AOP技术简化了Java应用的开发流程,降低了组件间的耦合度,支持事务管理和多种框架的无缝集成,极大提升了开发效率和代码质量。Spring 5引入了响应式编程等新特性,进一步增强了框架的功能性和灵活性。
65 0
|
2月前
|
安全 Java 测试技术
Java开发必读,谈谈对Spring IOC与AOP的理解
Spring的IOC和AOP机制通过依赖注入和横切关注点的分离,大大提高了代码的模块化和可维护性。IOC使得对象的创建和管理变得灵活可控,降低了对象之间的耦合度;AOP则通过动态代理机制实现了横切关注点的集中管理,减少了重复代码。理解和掌握这两个核心概念,是高效使用Spring框架的关键。希望本文对你深入理解Spring的IOC和AOP有所帮助。
53 0

热门文章

最新文章