当前位置： 首页 > news >正文

JMS 之 Active MQ 的spring整合

news 来源：原创 2024/5/18 13:53:22

一、与spring整合实现ptp的同步接收消息

pom.xml:

   <!-- https://mvnrepository.com/artifact/org.springframework/spring-jms -->
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-jms</artifactId>
        <version>4.3.7.RELEASE</version>
    </dependency>
  <!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-pool -->
    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-pool</artifactId>
        <version>5.9.0</version>
    </dependency>

spring-jms.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:jms="http://www.springframework.org/schema/jms"
xmlns:amq="http://activemq.apache.org/schema/core"
xsi:schemaLocation="http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd">

<!-- ActiveMQConnectionFactory就是JMS中负责创建到ActiveMQ连接的工厂类 -->
<bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory" >
     <property name="brokerURL" value="tcp://192.168.0.224:61616"/>  
</bean>
<!-- 创建连接池 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">  
    <property name="connectionFactory" ref="connectionFactory"/>  
    <property name="maxConnections" value="10"/>  
</bean>  
<!-- Spring为我们提供了多个ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory -->
<bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">  
    <property name="targetConnectionFactory" ref="pooledConnectionFactory"/>  
</bean> 

<!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->  
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
    <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  
    <property name="connectionFactory" ref="cachingConnectionFactory"/>  
</bean> 

<!--这个是队列目的地，点对点的-->  
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">  
    <constructor-arg index="0" value="spring-queue"/> 
</bean>   

</beans>

   ConnectionFactory是用于产生到JMS服务器的链接的，Spring为我们提供了多个ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory。SingleConnectionFactory对于建立JMS服务器链接的请求会一直返回同一个链接，并且会忽略Connection的close方法调用。CachingConnectionFactory继承了SingleConnectionFactory，所以它拥有SingleConnectionFactory的所有功能，同时它还新增了缓存功能，它可以缓存Session、MessageProducer和MessageConsumer。这里我们使用CachingConnectionFactory来作为示例。

消息生产者：

package com.jalja.org.jms.spring;

import javax.jms.Destination;
import javax.jms.JMSException;
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;

public class SpringJmsSend {
    public static void main(String[] args) {
        ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
        JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
        Destination queueDestination=(Destination) context.getBean("queueDestination");
        jmsTemplate.send(queueDestination, new MessageCreator(){
            @Override
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage("Hello spring JMS");
            }
        });
    }
}

消费者：

package com.jalja.org.jms.spring;

import javax.jms.Destination;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jms.core.JmsTemplate;

public class SpringJmsReceive {
public static void main(String[] args) {
    ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
    JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
    Destination queueDestination=(Destination) context.getBean("queueDestination");
    String msg=(String) jmsTemplate.receiveAndConvert(queueDestination);
    System.out.println(msg);
　　}
}

二、PTP的异步调用

 我们在spring中直接配置异步接收消息的监听器，这样就相当于在spring中配置了消费者，在接受消息的时候就不必要启动消费者了。

spring-jms.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:jms="http://www.springframework.org/schema/jms"
xmlns:amq="http://activemq.apache.org/schema/core"
xsi:schemaLocation="http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd">

<!-- ActiveMQConnectionFactory就是JMS中负责创建到ActiveMQ连接的工厂类 -->
<bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory" >
     <property name="brokerURL" value="tcp://192.168.0.224:61616"/>  
</bean>
<!-- 创建连接池 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">  
    <property name="connectionFactory" ref="connectionFactory"/>  
    <property name="maxConnections" value="10"/>  
</bean>  
<!-- Spring为我们提供了多个ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory -->
<bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">  
    <property name="targetConnectionFactory" ref="pooledConnectionFactory"/>  
</bean> 

<!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->  
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
    <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  
    <property name="connectionFactory" ref="cachingConnectionFactory"/>  
</bean> 

<!--这个是队列目的地，点对点的-->  
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">  
    <constructor-arg index="0" value="spring-queue"/> 
</bean>   

<!-- 消息监听器 -->  
<bean id="myMessageListener" class="com.jalja.org.jms.spring.yb.MyMessageListener"/>  

<!-- 消息监听容器 -->  
<bean id="jmsContainer"  class="org.springframework.jms.listener.DefaultMessageListenerContainer">  
    <property name="connectionFactory" ref="cachingConnectionFactory" />  
    <property name="destination" ref="queueDestination" />  
    <property name="messageListener" ref="myMessageListener" />  
</bean> 
</beans>

　　生产者往指定目的地Destination发送消息后，接下来就是消费者对指定目的地的消息进行消费了。那么消费者是如何知道有生产者发送消息到指定目的地Destination了呢？这是通过Spring为我们封装的消息监听容器MessageListenerContainer实现的，它负责接收信息，并把接收到的信息分发给真正的MessageListener进行处理。每个消费者对应每个目的地都需要有对应的MessageListenerContainer。对于消息监听容器而言，除了要知道监听哪个目的地之外，还需要知道到哪里去监听，也就是说它还需要知道去监听哪个JMS服务器，这是通过在配置MessageConnectionFactory的时候往里面注入一个ConnectionFactory来实现的。所以我们在配置一个MessageListenerContainer的时候有三个属性必须指定，一个是表示从哪里监听的ConnectionFactory；一个是表示监听什么的Destination；一个是接收到消息以后进行消息处理的MessageListener。Spring一共为我们提供了两种类型的MessageListenerContainer，SimpleMessageListenerContainer和DefaultMessageListenerContainer。
　　SimpleMessageListenerContainer：SimpleMessageListenerContainer会在一开始的时候就创建一个会话session和消费者Consumer，并且会使用标准的JMS MessageConsumer.setMessageListener()方法注册监听器让JMS提供者调用监听器的回调函数。它不会动态的适应运行时需要和参与外部的事务管理。兼容性方面，它非常接近于独立的JMS规范，但一般不兼容Java EE的JMS限制。

　　DefaultMessageListenerContainer：在大多数情况下我们还是使用的DefaultMessageListenerContainer，跟SimpleMessageListenerContainer相比，DefaultMessageListenerContainer会动态的适应运行时需要，并且能够参与外部的事务管理。它很好的平衡了对JMS提供者要求低、先进功能如事务参与和兼容Java EE环境。

消息生产者：

    public static void main(String[] args) {
        ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
        JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
        Destination queueDestination=(Destination) context.getBean("queueDestination");
        System.out.println("异步调用执行开始");
        jmsTemplate.send(queueDestination, new MessageCreator(){
            @Override
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage("Hello spring JMS");
            }
        });
        System.out.println("异步调用执行结束");
    }

消息监听器：MyMessageListener

public class MyMessageListener implements MessageListener{
    @Override
    public void onMessage(Message message) {
        TextMessage msg= (TextMessage) message;
        try {
            System.out.println("你好："+msg.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

启动消息生产者  监听器的执行结果是：

异步调用执行开始
异步调用执行结束
你好：Hello spring JMS

三、发布订阅 同步接收

spring-jms.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:jms="http://www.springframework.org/schema/jms"
xmlns:amq="http://activemq.apache.org/schema/core"
xsi:schemaLocation="http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd">

<!-- ActiveMQConnectionFactory就是JMS中负责创建到ActiveMQ连接的工厂类 -->
<bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory" >
     <property name="brokerURL" value="tcp://192.168.0.224:61616"/>  
</bean>
<!-- 创建连接池 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">  
    <property name="connectionFactory" ref="connectionFactory"/>  
    <property name="maxConnections" value="10"/>  
</bean>  
<!-- Spring为我们提供了多个ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory -->
<bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">  
    <property name="targetConnectionFactory" ref="pooledConnectionFactory"/>  
</bean> 

<!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->  
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
    <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  
    <property name="connectionFactory" ref="cachingConnectionFactory"/>  
</bean> 

<!--这个是队列目的地，发布订阅-->  
<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">  
    <constructor-arg index="0" value="spring-Topic"/> 
</bean>   
</beans>

生产者：

    public static void main(String[] args) {
        ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
        JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
        Destination topicDestination=(Destination) context.getBean("topicDestination");
        jmsTemplate.send(topicDestination, new MessageCreator(){
            @Override
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage("Hello spring JMS topicDestination");
            }
        });
    }

消费者：

public class SpringJmsSubscriber {
    public static void main(String[] args) {
        ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
        JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
        Destination topicDestination=(Destination) context.getBean("topicDestination");
        String msg=(String) jmsTemplate.receiveAndConvert(topicDestination);
        System.out.println(msg);
    }
}

 

转载于:https://www.cnblogs.com/jalja/p/7067741.html

相关文章：

  • Java实现排序算法(二)
  • vue项目实战爬坑小记001
  • Java实现排序算法(三)
  • Java通信编程之Socket入门
  • 回车提交表单
  • 数据库的查询优化技术
  • 微信jssdk分享功能，jssdk成功调用，分享内容自定义失败
  • 用VS2005制造WEB安装程序
  • Java对象实例化顺序
  • 非阻塞模式下，虽然connect出错，但是getsockopt取得的错误却是0的问题
  • Java面试之判断对错
  • PL/SQL
  • 数据库系统优化--业务逻辑设计优化
  • 【WPF】DispatcherFrame 是个啥玩意儿
  • 技术路线的选择重要但不具有决定性
  • DataBase in Android
  • iOS 颜色设置看我就够了
  • Javascript基础之Array数组API
  • JavaScript设计模式系列一：工厂模式
  • mac修复ab及siege安装
  • miniui datagrid 的客户端分页解决方案 - CS结合
  • MySQL的数据类型
  • MySQL主从复制读写分离及奇怪的问题
  • Promise初体验
  • Python进阶细节
  • 小程序开发中的那些坑
  • 学习Vue.js的五个小例子
  • 继 XDL 之后，阿里妈妈开源大规模分布式图表征学习框架 Euler ...
  • ​MySQL主从复制一致性检测
  • #我与Java虚拟机的故事#连载07：我放弃了对JVM的进一步学习
  • #中国IT界的第一本漂流日记 传递IT正能量# 【分享得“IT漂友”勋章】
  • （1）STL算法之遍历容器
  • （入门自用）--C++--抽象类--多态原理--虚表--1020
  • （十）DDRC架构组成、效率Efficiency及功能实现
  • ..回顾17，展望18
  • .dat文件写入byte类型数组_用Python从Abaqus导出txt、dat数据
  • .mysql secret在哪_MySQL如何使用索引
  • .Net Memory Profiler的使用举例
  • .net 使用$.ajax实现从前台调用后台方法(包含静态方法和非静态方法调用)
  • .NET/C# 阻止屏幕关闭，阻止系统进入睡眠状态
  • .NET连接MongoDB数据库实例教程
  • @ComponentScan比较
  • @select 怎么写存储过程_你知道select语句和update语句分别是怎么执行的吗？
  • @Transactional注解下，循环取序列的值，但得到的值都相同的问题
  • [ CTF ]【天格】战队WriteUp- 2022年第三届“网鼎杯”网络安全大赛（青龙组）
  • [2544]最短路 （两种算法）（HDU）
  • [android] 看博客学习hashCode()和equals()
  • [BZOJ1060][ZJOI2007]时态同步 树形dp
  • [BZOJ1877][SDOI2009]晨跑[最大流+费用流]
  • [IE编程] 如何获得IE版本号
  • [IE技巧] 使IE8以单进程的模式运行
  • [JavaEE]线程的状态与安全
  • [jQuery]10 Things I Learned from the jQuery Source
  • [JS入门到进阶] 哎，被vite小坑了一波，大家记得配置build.cssTarget为‘chrome61‘
  • [k8s系列]：kubernetes·概念入门