ActiveMQ专题7 —— Spring整合ActiveMQ

喜欢ヅ旅行 2022-11-05 12:57 252阅读 0赞

## 环境准备 ##

*  启动的ActiveMQ服务
 *  JDK1.8+
 *  IDEA或Eclipse
 *  Maven环境
 *  Spring环境依赖

核心依赖

<dependency>
          <groupId>org.apache.activemq</groupId>
          <artifactId>activemq-all</artifactId>
          <version>5.15.9</version>
        </dependency>
      
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-pool</artifactId>
            <version>5.15.10</version>
        </dependency>
        
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-jms</artifactId>
            <version>5.2.1.RELEASE</version>
        </dependency>

全部pom依赖，参考

<dependencies>
       
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-all</artifactId>
            <version>5.10.0</version>
        </dependency>
        
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.10.1</version>
        </dependency>
        
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-pool</artifactId>
            <version>5.15.10</version>
        </dependency>
        
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-jms</artifactId>
            <version>5.2.1.RELEASE</version>
        </dependency>
        
        <dependency>
            <groupId>org.apache.xbean</groupId>
            <artifactId>xbean-spring</artifactId>
            <version>4.15</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-aop</artifactId>
            <version>5.2.1.RELEASE</version>
        </dependency>
        
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-core</artifactId>
            <version>4.3.23.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
            <version>4.3.23.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-aop</artifactId>
            <version>4.3.23.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-orm</artifactId>
            <version>4.3.23.RELEASE</version>
        </dependency>
    </dependencies>

## 配置文件 ##

<?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:context="http://www.springframework.org/schema/context"
           xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context https://www.springframework.org/schema/context/spring-context.xsd">
    
        
        <context:component-scan base-package="com.huazai.activemq.spring"></context:component-scan>
    
        
        <bean id="connectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
            <property name="connectionFactory">
                
                <bean class="org.apache.activemq.spring.ActiveMQConnectionFactory">
                    <property name="brokerURL" value="tcp://7.98.163.118:61616"/>
                </bean>
            </property>
            
            <property name="maxConnections" value="100"/>
        </bean>
    
        
        <bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">
            
            <constructor-arg index="0" value="spring-active-queue"/>
        </bean>
    
        
        <bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">
            
            <constructor-arg index="0" value="spring-active-topic"/>
        </bean>
    
        
        <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
            
            <property name="connectionFactory" ref="connectionFactory"/>
            
            <property name="defaultDestination" ref="destinationTopic"/>
            
            <property name="messageConverter">
                <bean class="org.springframework.jms.support.converter.SimpleMessageConverter"/>
            </property>
        </bean>
    </beans>

## 生产者Queue ##

队列生产者代码：

package com.huazai.activemq.spring.queue;
    
    import org.apache.xbean.spring.context.ClassPathXmlApplicationContext;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.ApplicationContext;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.stereotype.Service;
    
    import java.util.Scanner;
    
    /**
     * @author pyh
     * @date 2021/01/11 23:39
     */
    @Service
    public class SpringMQProducer {
        
        static final Scanner input = new Scanner(System.in);
    
        @Autowired
        private final JmsTemplate jmsTemplate;
    
        @Autowired
        public SpringMQProducer(JmsTemplate jmsTemplate) {
        
            this.jmsTemplate = jmsTemplate;
        }
    
        public static void main(String[] args) {
        
            ApplicationContext applicationContext = new ClassPathXmlApplicationContext("applicationContext.xml");
            SpringMQProducer springMQ_producer = applicationContext.getBean(SpringMQProducer.class);
            JmsTemplate jmsTemplate = springMQ_producer.jmsTemplate;
            // 发送消息
            jmsTemplate.send(session -> session.createTextMessage("***Spring和ActiveMQ的整合case....."));
            System.out.println("********send task over");
            while(true) {
        
                System.out.println("\n请输入要发送的内容");
                final String msgText = input.next();
                // 输入exit推出程序
                if ("exit".equals(msgText)) {
        
                    System.exit(-1);
                }
                if (msgText.startsWith("auto:")) {
        
                    int amount;
                    try {
        
                        amount = Integer.parseInt(msgText.substring(5));
                    } catch (NumberFormatException e) {
        
                        System.out.println("auto指令格式错误，正确的应该是：auto:number");
                        continue;
                    }
                    System.out.println("请稍等...");
                    for (int i = 1; i <= amount; i++) {
        
                        final int ii = i;
                        jmsTemplate.send(session -> session.createTextMessage(ii + ""));
                    }
                    System.out.println(msgText + "\t->|\t指令执行完成");
                } else {
        
                    jmsTemplate.send(session -> session.createTextMessage(msgText));
                    System.out.println(msgText + "\t->|\t发送完成");
                }
            }
        }
    }

队列消费者代码：

启动消费者和订阅者，结果如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW96aGVnYWE_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][20210306182101804.png]

## 生产者Topic ##

增加核心配置

<bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">
            <constructor-arg index="0" value="spring-active-topic"></constructor-arg>
        </bean>

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW96aGVnYWE_size_16_color_FFFFFF_t_70 1]消费者和生产者代码不需要改变  
 就 这 么 简 简 单 单 \\color\{red\}就这么简简单单 就这么简简单单

## 监听器实现消费者不启动直接消费消息 ##

applicationContext.xml 监听器配置：

<bean id="defaultMessageListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        
        <property name="connectionFactory" ref="connectionFactory"/>
        
        <property name="destination" ref="destinationTopic"/>
        
        <property name="messageListener" ref="myMessageListener"/>
    </bean>

自定义监听器：

package com.huazai.activemq.spring.listener;
    
    import org.springframework.stereotype.Component;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    /**
     * @author pyh
     * @date 2021/1/12 0:14
     */
    @Component
    public class MyMessageListener implements MessageListener {
        
        @Override
        public void onMessage(Message message) {
        
            if (message instanceof TextMessage) {
        
                TextMessage textMessage = (TextMessage) message;
                try {
        
                    System.out.println("消费者收到的消息" + textMessage.getText());
                } catch (JMSException e) {
        
                    e.printStackTrace();
                }
            }
        }
    }

增加了监听器之后，在原来代码基础上

1.  只启动队列生产者  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW96aGVnYWE_size_16_color_FFFFFF_t_70 2]  
    以上案例实现了不启动消费者通过监听器实现queue和topic消息消费。

## 连更文章 ##

[ActiveMQ专题0 —— 前言说明][ActiveMQ_0 _]  
[ActiveMQ专题1 —— 入门概述][ActiveMQ_1 _]  
[ActiveMQ专题2 —— ActiveMQ下载和安装（Linux版）][ActiveMQ_2 _ ActiveMQ_Linux]  
[ActiveMQ专题3 ——Java编码实现ActiveMQ通讯（Queue）][ActiveMQ_3 _Java_ActiveMQ_Queue]  
[ActiveMQ专题4 ——ActiveMQ专题4 ——Java编码实现ActiveMQ通讯（Topic）][ActiveMQ_4 _ActiveMQ_4 _Java_ActiveMQ_Topic]  
[ActiveMQ专题5 ——JMS规范和落地产品][ActiveMQ_5 _JMS]  
[ActiveMQ专题6 ——ActiveMQ的Broker][ActiveMQ_6 _ActiveMQ_Broker]  
[ActiveMQ专题7 —— Spring整合ActiveMQ][ActiveMQ_7 _ Spring_ActiveMQ]  
[ActiveMQ专题8 —— SpringBoot整合ActiveMQ][ActiveMQ_8 _ SpringBoot_ActiveMQ]  
[ActiveMQ专题9 —— ActiveMQ的传输协议][ActiveMQ_9 _ ActiveMQ]  
[ActiveMQ专题10 —— ActiveMQ的存储和持久化][ActiveMQ_10 _ ActiveMQ]  
[ActiveMQ专题11 —— ActiveMQ的存储和持久化2 （续篇）][ActiveMQ_11 _ ActiveMQ_2]  
[ActiveMQ专题12—— ActiveMQ之zookeeper集群][ActiveMQ_12_ ActiveMQ_zookeeper]  
[ActiveMQ专题13—— ActiveMQ高级特性][ActiveMQ_13_ ActiveMQ]  
[ActiveMQ专题14—— ActiveMQ高级特性2][ActiveMQ_14_ ActiveMQ_2]  
$\\color\{red\} 完结！后续推上有关消息队列的面试题目 $

## 小总结 ##

知 道 的 越 多 ， 不 知 道 的 越 多 ， 希 望 对 你 有 帮 助 ！ \\color\{red\}知道的越多，不知道的越多，希望对你有帮助！ 知道的越多，不知道的越多，希望对你有帮助！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW96aGVnYWE_size_16_color_FFFFFF_t_70]: /images/20221023/ff52703557c14a2f97e96f2a8aa61565.png
[20210306182101804.png]: /images/20221023/637287d0658147e4a2bc2be59a60bd58.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW96aGVnYWE_size_16_color_FFFFFF_t_70 1]: /images/20221023/5dad376002524d81837b14764eb9ee2d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW96aGVnYWE_size_16_color_FFFFFF_t_70 2]: /images/20221023/4ff4378a3c064deb892d0cd998f0de61.png
[ActiveMQ_0 _]: https://blog.csdn.net/xiaozhegaa/article/details/114364355
[ActiveMQ_1 _]: https://blog.csdn.net/xiaozhegaa/article/details/114373971
[ActiveMQ_2 _ ActiveMQ_Linux]: https://blog.csdn.net/xiaozhegaa/article/details/114375436
[ActiveMQ_3 _Java_ActiveMQ_Queue]: https://blog.csdn.net/xiaozhegaa/article/details/114382174
[ActiveMQ_4 _ActiveMQ_4 _Java_ActiveMQ_Topic]: https://blog.csdn.net/xiaozhegaa/article/details/114383509
[ActiveMQ_5 _JMS]: https://blog.csdn.net/xiaozhegaa/article/details/114388827
[ActiveMQ_6 _ActiveMQ_Broker]: https://blog.csdn.net/xiaozhegaa/article/details/114449488
[ActiveMQ_7 _ Spring_ActiveMQ]: https://blog.csdn.net/xiaozhegaa/article/details/114450231
[ActiveMQ_8 _ SpringBoot_ActiveMQ]: https://blog.csdn.net/xiaozhegaa/article/details/114451603
[ActiveMQ_9 _ ActiveMQ]: https://blog.csdn.net/xiaozhegaa/article/details/114452474
[ActiveMQ_10 _ ActiveMQ]: https://blog.csdn.net/xiaozhegaa/article/details/114457556
[ActiveMQ_11 _ ActiveMQ_2]: https://blog.csdn.net/xiaozhegaa/article/details/114461996
[ActiveMQ_12_ ActiveMQ_zookeeper]: https://blog.csdn.net/xiaozhegaa/article/details/114464565
[ActiveMQ_13_ ActiveMQ]: https://blog.csdn.net/xiaozhegaa/article/details/114494510
[ActiveMQ_14_ ActiveMQ_2]: https://blog.csdn.net/xiaozhegaa/article/details/114500737