ActiveMQ —— Spring 整合 ActiveMQ

柔情只为你懂 2023-02-11 07:28 134阅读 0赞

前文

消息中间件 —— 简介

ActiveMQ 下载、安装

ActiveMQ —— Java 连接 ActiveMQ（点对点）

ActiveMQ —— Java 连接 ActiveMQ（发布订阅 Topic）

ActiveMQ —— Broker

文章目录

  • 前文
  • 添加依赖
  • Spring 配置文件
  • 队列
    • 生产者
    • 消费者
  • 主题
    • 修改 applicationContext.xml 文件
  • 在 Spring 里面实现消费者不启动，直接通过配置监听完成

添加依赖

  1. <dependency>
  2.   <groupId>com.fasterxml.jackson.core</groupId>
  3.   <artifactId>jackson-databind</artifactId>
  4.   <version>2.9.5</version>
  5. </dependency>
  7. <!-- ActiveMQ 对 JMS 支持，整合 Spring 和 ActiveMQ -->
  8. <dependency>
  9.   <groupId>org.springframework</groupId>
  10.   <artifactId>spring-jms</artifactId>
  11.   <version>4.3.23.RELEASE</version>
  12. </dependency>
  14. <!-- ActiveMQ 所需要的 pool 包配置 -->
  15. <dependency>
  16.   <groupId>org.apache.activemq</groupId>
  17.   <artifactId>activemq-pool</artifactId>
  18.   <version>5.15.9</version>
  19. </dependency>
  21. <!-- Spring AOP 等相关 Jar 包 -->
  22. <dependency>
  23.   <groupId>org.springframework</groupId>
  24.   <artifactId>spring-core</artifactId>
  25.   <version>4.3.25.RELEASE</version>
  26. </dependency>

这里的 Spring-core 有版本问题，使用 5.0+ 的整合 ActiveMQ 会报错，使用 4.0 + 的就不会

Spring 配置文件

  1. <?xml version="1.0" encoding="UTF-8" ?>
  2. <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.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/beans">
  4.     <!-- 开启包的自动扫描 -->
  5.     <context:component-scan base-package="com.java.elasticsearch.activemq"/>
  7.     <!-- 配置生产者 -->
  8.     <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
  9.         <property name="connectionFactory">
  10.             <!-- 真正可以产生 Connection 的 ConnectionFactory，由对应的 JMS 服务厂商提供 -->
  11.             <bean class="org.apache.activemq.ActiveMQConnectionFactory">
  12.                 <property name="brokerURL" value="tcp://localhost:61616"/>
  13.             </bean>
  14.         </property>
  15.         <!-- 最大连接数 -->
  16.         <property name="maxConnections" value="100"/>
  17.     </bean>
  19.     <!-- 这个是队列的目的地，点对点的 -->
  20.     <bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">
  21.         <!-- 队列名称 -->
  22.         <constructor-arg index="0" value="spring-active-queue"/>
  23.     </bean>
  25.     <!-- Spring 提供的 JMS 工具类，它可以进行消息发送，接收等 -->
  26.     <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
  27.         <property name="connectionFactory" ref="jmsFactory"/>
  28.         <property name="defaultDestination" ref="destinationQueue"/>
  29.         <property name="messageConverter">
  30.             <bean class="org.springframework.jms.support.converter.SimpleMessageConverter"/>
  31.         </property>
  32.      </bean>
  34. </beans>

队列

生产者

  1. package com.java.elasticsearch.activemq.service;
  3. import org.springframework.beans.factory.annotation.Autowired;
  4. import org.springframework.context.ApplicationContext;
  5. import org.springframework.context.support.ClassPathXmlApplicationContext;
  6. import org.springframework.jms.core.JmsTemplate;
  7. import org.springframework.jms.core.MessageCreator;
  8. import org.springframework.stereotype.Service;
  9. import javax.jms.JMSException;
  10. import javax.jms.Message;
  11. import javax.jms.Session;
  12. import javax.jms.TextMessage;
  14. /** * @author Woo_home * @create 2020/5/22 14:09 */
  16. @Service
  17. public class SpringMQ_Produce { 
  19.     @Autowired
  20.     private JmsTemplate jmsTemplate;
  22.     public static void main(String[] args){ 
  23.         ApplicationContext ctx = new ClassPathXmlApplicationContext("applicationContext.xml");
  25.         // 相当于 new 一个 SpringMQ_Produce，但是这里用的是 Spring，所以这里就不用 new 了
  26.         SpringMQ_Produce produce = (SpringMQ_Produce) ctx.getBean("springMQ_Produce");
  28.         // 普通写法
  29.         /*produce.jmsTemplate.send(new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { TextMessage textMessage = session.createTextMessage("****** Spring 整合 ActiveMQ ******"); return textMessage; } });*/
  31.         // lambda 写法
  32.         produce.jmsTemplate.send((session) -> { 
  33.             TextMessage textMessage = session.createTextMessage("****** Spring 整合 ActiveMQ ******");
  34.             return textMessage;
  35.         });
  36.     }
  37. }

执行程序之前先启动 ActiveMQ
在这里插入图片描述
执行主程序
在这里插入图片描述
刷新 admin 页面，队列已成功添加进来
在这里插入图片描述

消费者

消费者端代码非常简单，只需要接收就行了

  1. package com.java.elasticsearch.activemq.service;
  3. import org.springframework.beans.factory.annotation.Autowired;
  4. import org.springframework.context.ApplicationContext;
  5. import org.springframework.context.support.ClassPathXmlApplicationContext;
  6. import org.springframework.jms.core.JmsTemplate;
  7. import org.springframework.stereotype.Service;
  8. import javax.jms.TextMessage;
  10. /** * @author Woo_home * @create 2020/5/22 14:09 */
  12. @Service
  13. public class SpringMQ_Consumer { 
  15.     @Autowired
  16.     private JmsTemplate jmsTemplate;
  18.     public static void main(String[] args){ 
  19.         ApplicationContext ctx = new ClassPathXmlApplicationContext("applicationContext.xml");
  21.         // 相当于 new 一个 SpringMQ_Consumer，但是这里用的是 Spring，所以这里就不用 new 了
  22.         SpringMQ_Consumer consumer = (SpringMQ_Consumer) ctx.getBean("springMQ_Consumer");
  24.         String retValue  = (String) consumer.jmsTemplate.receiveAndConvert();
  26.         System.out.println("****** 消费者收到消息 : " + retValue);
  27.     }
  28. }

执行程序
在这里插入图片描述
刷新 admin 页面
在这里插入图片描述

主题

修改 applicationContext.xml 文件

只需要将以下三处修改为 Topic 即可
在这里插入图片描述
生产者与消费者的代码不用修改，直接运行，要是你想修改的话，可以修改如下内容
在这里插入图片描述
在这里插入图片描述
因为这里使用的是 Topic，我们先启动消费端，此时的消费端还没收到信息
在这里插入图片描述
在这里插入图片描述
再启动生产端
在这里插入图片描述
在这里插入图片描述
此时，消费端接收到生产者发来的信息
在这里插入图片描述

在 Spring 里面实现消费者不启动，直接通过配置监听完成

修改 applicationContext.xml 文件

  1. <?xml version="1.0" encoding="UTF-8" ?>
  2. <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.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/beans">
  4.     <!-- 开启包的自动扫描 -->
  5.     <context:component-scan base-package="com.java.elasticsearch.activemq"/>
  7.     <!-- 配置生产者 -->
  8.     <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
  9.         <property name="connectionFactory">
  10.             <!-- 真正可以产生 Connection 的 ConnectionFactory，由对应的 JMS 服务厂商提供 -->
  11.             <bean class="org.apache.activemq.ActiveMQConnectionFactory">
  12.                 <property name="brokerURL" value="tcp://localhost:61616"/>
  13.             </bean>
  14.         </property>
  15.         <!-- 最大连接数 -->
  16.         <property name="maxConnections" value="100"/>
  17.     </bean>
  19.     <!-- 这个是队列的目的地，点对点的 -->
  20.     <bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">
  21.         <!-- 队列名称 -->
  22.         <constructor-arg index="0" value="spring-active-queue"/>
  23.     </bean>
  25.     <!-- 这个是队列的目的地，主题的 -->
  26.     <bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">
  27.         <!-- 队列名称 -->
  28.         <constructor-arg index="0" value="spring-active-topic"/>
  29.     </bean>
  31.     <!-- Spring 提供的 JMS 工具类，它可以进行消息发送，接收等 -->
  32.     <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
  33.         <property name="connectionFactory" ref="jmsFactory"/>
  34.         <property name="defaultDestination" ref="destinationTopic"/>
  35.         <property name="messageConverter">
  36.             <bean class="org.springframework.jms.support.converter.SimpleMessageConverter"/>
  37.         </property>
  38.      </bean>
  40.     <!-- 配置监听程序 -->
  41.     <bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  42.         <property name="connectionFactory" ref="jmsFactory"/>
  43.         <property name="destination" ref="destinationTopic"/>
  44.         <!-- public class MyMessageListener implements MessageListener {} -->
  45.         <property name="messageListener" ref="myMessageListener"/>
  46.     </bean>
  48. </beans>

配置监听
在这里插入图片描述
这里还需要一个组件

  1. package com.java.elasticsearch.activemq.service;
  3. import org.springframework.stereotype.Component;
  5. import javax.jms.JMSException;
  6. import javax.jms.Message;
  7. import javax.jms.MessageListener;
  8. import javax.jms.TextMessage;
  10. /** * @author Woo_home * @create 2020/5/22 14:59 */
  12. @Component
  13. public class MyMessageListener implements MessageListener { 
  14.     @Override
  15.     public void onMessage(Message message) { 
  16.         if (null != message && message instanceof TextMessage) { 
  17.             TextMessage textMessage = (TextMessage) message;
  18.             try { 
  19.                 System.out.println(textMessage.getText());
  20.             } catch (JMSException e) { 
  21.                 e.printStackTrace();
  22.             }
  23.         }
  24.     }
  25. }

先删除 Topic
在这里插入图片描述
启动生产者，可以发现，消费者马上就可以监听到消费者发送的消息
在这里插入图片描述
在这里插入图片描述

完整代码已上传至码云 代码下载地址

发表评论

表情：
评论列表 （有 0 条评论，134人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Spring 整合 ActiveMQ

    Spring 是J2EE 最重要的框架，ActiveMQ 是Jms的框架，用于两个程序、系统中的异步通信，两者的用途都挺广泛。上一篇博文介绍的是发布-订阅形式，今次以点-点形式

    水深无声水深无声/ 0 赞/ 329 阅读