微服务消息中间件 ActiveMQ

向右看齐 2023-01-04 11:59 229阅读 0赞

### 微服务 消息中间件 ActiveMQ ###

*  微服务 消息中间件 ActiveMQ
 *   *  一、ActiveMQ概念
     *   *  1、MQ概述
         *  2、JMS概念
         *   *  JMS概念
             *  JMS消息模型
             *  JMS编程API
     *  二、ActiveMQ下载安装
     *   *  1、官网下载
         *  2、Linux环境安装
         *  3、Windows环境安装
         *  4、控制台
     *  三、JMS消息组成
     *   *  1、jms消息组成
         *  2、jms消息头
         *  3、jms消息体
         *  4、jms消息属性
     *  四、ActiveMQ特性
     *   *  1、消息持久化
         *  2、消息事务
         *  3、确认机制
         *  4、消息投递方式
         *  5、死信队列
     *  五、ActiveMQ操作案例
     *  六、原生jms操作ActiveMQ案例
     *  七、Spring整合ActiveMQ案例
     *  八、SpringBoot整合ActiveMQ案例

# 微服务 消息中间件 ActiveMQ #

## 一、ActiveMQ概念 ##

### 1、MQ概述 ###

MQ就是消息中间件。
    面向消息的中间件（message-oriented middleware）MOM能够很好的解决以上问题。
    是指利用高效可靠的消息传递机制与平台无关的数据交流，并基于数据通信来进行分布式系统的集成。
    通过提供消息传递和消息排队模型在分布式环境下提供应用解耦，弹性伸缩，冗余存储、流量削峰，异步通信，数据同步等功能。
    
    大致的过程是这样的：发送者把消息发送给消息服务器，消息服务器将消息存放在若干队列/主题topic中，在合适的时候，消息服务器回将消息转发给接受者。
    在这个过程中，发送和接收是异步的，也就是发送无需等待，而且发送者和接受者的生命周期也没有必然的关系；尤其在发布pub/订阅sub模式下，也可以完成一对多的通信，即让一个消息有多个接受者。
    
    官网介绍
    ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。我们在本次课程中介绍 ActiveMQ的使用。
    总结一句话ActiveMQ就是消息中间件。

[微服务 MQ消息中间件][MQ]  
[消息中间件ActiveMQ课程笔记][ActiveMQ]

### 2、JMS概念 ###

#### JMS概念 ####

消息中间件利用高效可靠的消息传递机制进行平台无关的数据交流，并基于数据通信来进行分布式系统的集成。它可以在分布式环境下扩展进程间的通信。对于消息中间件，常见的角色大致也就有Producer（生产者）、Consumer（消费者）。
    
    消息队列中间件是分布式系统中重要的组件，主要解决应用解耦，异步消息，流量削锋等问题，实现高性能，高可用，可伸缩和最终一致性架构。
    
    JMS（Java Messaging Service）是Java平台上有关面向消息中间件的技术规范，它便于消息系统中的Java应用程序进行消息交换,并且通过提供标准的产生、发送、接收消息的接口简化企业应用的开发。
    
    JMS本身只定义了一系列的接口规范，是一种与厂商无关的 API，用来访问消息收发系统。它类似于JDBC(java Database Connectivity)：这里，JDBC 是可以用来访问许多不同关系数据库的 API，而 JMS则提供同样与厂商无关的访问方法，以访问消息收发服务。

#### JMS消息模型 ####

消息中间件一般有两种传递模式：点对点模式(P2P)和发布-订阅模式(Pub/Sub)。
    (1) P2P (Point to Point) 点对点模型（Queue队列模型）
    (2) Publish/Subscribe(Pub/Sub) 发布/订阅模型(Topic主题模型)
    
    一般理解点对点是队列形式，发布订阅类似于微信公众号。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70]![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 1]

#### JMS编程API ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 2]![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 3]

## 二、ActiveMQ下载安装 ##

### 1、官网下载 ###

官网地址：[http://activemq.apache.org/][http_activemq.apache.org]  
Components -->ActiveMQ5 --> Download  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 4]

### 2、Linux环境安装 ###

第一步：安装 jdk（略）
    第二步：把 activemq的压缩包（apache-activemq-5.14.5-bin.tar.gz）上传到 linux 系统
    第三步：解压缩压缩包
    tar -zxvf apache-activemq-5.14.5-bin.tar.gz
    第四步：进入apache-activemq-5.14.5的bin目录
    cd apache-activemq-5.14.5/bin
    第五步：启动 activemq
    ./activemq start (执行2次：第一次：生成配置信息；第二次：启动
    第六步：停止activemq：
    ./activemq stop

### 3、Windows环境安装 ###

直接解压缩，到apache-activemq-5.15.14\bin\win64目录下执行activemq.bat

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 5]

### 4、控制台 ###

访问地址：[http://localhost:8161/admin/queues.jsp][http_localhost_8161_admin_queues.jsp]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 6]![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 7]

Number Of Consumers  消费者 
    Number Of Pending Messages 等待消费的消息 
    Messages Enqueued 进入队列的消息 
    Messages Dequeued 出了队列的消息 
    Operations  操作 
    Purge 净化
    
    当有一个消息进入这个队列时，等待消费的消息是1，进入队列的消息是1。当消息消费后，等待消费的消息是0，进入队列的消息是1，出队列的消息是1.在来一条消息时，等待消费的消息是1，进入队列的消息就是2.以此类推，进入队列的消息和出队列的消息是池子，等待消费的消息是水流。

## 三、JMS消息组成 ##

### 1、jms消息组成 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 8]

### 2、jms消息头 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 9]

### 3、jms消息体 ###

在消息体中，JMS API定义了五种类型的消息格式，让我们可以以不同的形式发送和接受消息，并提供
    了对已有消息格式的兼容。不同的消息类型如下：
    JMS 定义了五种不同的消息正文格式，以及调用的消息类型，允许你发送并接收一些不同形式的数据，提供现有消息格式的一些级别的兼容性。
    · TextMessage--一个字符串对象 *
    · MapMessage--一套名称-值对
    · ObjectMessage--一个序列化的 Java 对象 *
    · BytesMessage--一个字节的数据流 *
    · StreamMessage -- Java原始值的数据流

### 4、jms消息属性 ###

我们可以给消息设置自定义属性，这些属性主要是提供给应用程序的。对于实现消息过滤功能，消息属性非常有用，JMS API定义了一些标准属性，JMS服务提供者可以选择性的提供部分标准属性.
    message.setStringProperty("Property",Property); //自定义属性

## 四、ActiveMQ特性 ##

### 1、消息持久化 ###

消息持久化是保证消息不丢失的重要方式！！！
    ActiveMQ提供了以下三种的消息存储方式：
    （1） Memory 消息存储-基于内存的消息存储。
    （2） 基于日志消息存储方式，KahaDB是ActiveMQ的默认日志存储方式，它提供了容量的提升和恢复能力。
    （3） 基于JDBC的消息存储方式-数据存储于数据库（例如：MySQL）中。

### 2、消息事务 ###

消息事务，是保证消息传递原子性的一个重要特征，和JDBC的事务特征类似。
    一个事务性发送，其中一组消息要么能够全部保证到达服务器，要么都不到达服务器。
    生产者、消费者与消息服务器直接都支持事务性；
    ActionMQ的事务主要偏向在生产者的应用。
    
    分为生产者事务与消费者事务。

### 3、确认机制 ###

注意：消息确认机制与事务机制是冲突的，只能选其中一种。所以演示消息确认前，先关闭事务。
    
    JMS消息只有在被确认之后，才认为已经被成功地消费了。消息的成功消费通常包含三个阶段：客户接收消息、客户处理消息和消息被确认。在事务性会话中，当一个事务被提交的时候，确认自动发生。在非事务性会话中，消息何时被确认取决于创建会话时的应答模式（acknowledgement mode）。该参数有以下三个可选值。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 10]

### 4、消息投递方式 ###

1、异步投递 vs 同步投递
    同步发送：
    消息生产者使用持久（Persistent）传递模式发送消息的时候，Producer.send() 方法会被阻塞，直到broker 发送一个确认消息给生产者(ProducerAck)，这个确认消息暗示broker已经成功接收到消息并把消息保存到二级存储中。
    
    异步发送:
    如果应用程序能够容忍一些消息的丢失，那么可以使用异步发送。异步发送不会在受到broker的确认之前一直阻塞 Producer.send方法。
    想要使用异步，在brokerURL中增加 jms.alwaysSyncSend=false&jms.useAsyncSend=true属性
    1）如果设置了alwaysSyncSend=true系统将会忽略useAsyncSend设置的值都采用同步 2）当
    alwaysSyncSend=false时，“NON_PERSISTENT”(非持久化)、事务中的消息将使用“异步发送” 3）当alwaysSyncSend=false时，如果指定了useAsyncSend=true，“PERSISTENT”类型的消息使用异步发送。如果useAsyncSend=false，“PERSISTENT”类型的消息使用同步发送。
    总结： 默认情况(alwaysSyncSend=false,useAsyncSend=false)，非持久化消息、事务内的消息均采用异步发送；对于持久化消息采用同步发送！！！

### 5、死信队列 ###

DLQ-Dead Letter Queue，死信队列，用来保存处理失败或者过期的消息
    
    出现以下情况时，消息会被重发：
    A transacted session is used and rollback() is called.
    A transacted session is closed before commit is called.
    A session is using CLIENT_ACKNOWLEDGE and Session.recover() is called.
    当一个消息被重发超过6(缺省为6次)次数时，会给broker发送一个"Poison ack"，这个消息被认为是apoison pill，这时broker会将这个消息发送到死信队列，以便后续处理。
    注意两点：
    1）缺省持久消息过期，会被送到DLQ，非持久消息不会送到DLQ
    2）缺省的死信队列是ActiveMQ.DLQ，如果没有特别指定，死信都会被发送到这个队列。
    可以通过配置文件(activemq.xml)来调整死信发送策略。

## 五、ActiveMQ操作案例 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 11]

## 六、原生jms操作ActiveMQ案例 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 12]jms\_producer

/** * 演示点对点模式 -- 消息生产者 * * @author zrj * @date 2021/1/9 * @since V1.0 **/
    public class PTP_Producer { 
    
        public static void main(String[] args) throws JMSException { 
    
            //1.创建连接工厂
            ConnectionFactory factory = new ActiveMQConnectionFactory( "tcp://127.0.0.1:61616" );
    
            //2.创建连接
            Connection connection = factory.createConnection();
    
            //3.打开连接
            connection.start();
    
            //4.创建session
            /** * 参数一：是否开启事务操作 * 参数二：消息确认机制 */
            Session session = connection.createSession( false, Session.AUTO_ACKNOWLEDGE );
    
            //5.创建目标地址（Queue:点对点消息，Topic：发布订阅消息）
            Queue queue = session.createQueue( "queue01" );
    
            //6.创建消息生产者
            MessageProducer producer = session.createProducer( queue );
    
            //7.创建消息
            //createTextMessage: 文本类型
            TextMessage textMessage = session.createTextMessage( "test message" );
    
            //8.发送消息
            producer.send( textMessage );
    
            System.out.println( "消息发送完成" );
    
            //9.释放资源
            session.close();
            connection.close();
        }
    }

/** * 演示发布订阅模式 -- 消息生产者 * * @author zrj * @date 2021/1/9 * @since V1.0 **/
    public class PS_Producer { 
    
        public static void main(String[] args) throws JMSException { 
    
            //1.创建连接工厂
            ConnectionFactory factory = new ActiveMQConnectionFactory( "tcp://127.0.0.1:61616" );
    
            //2.创建连接
            Connection connection = factory.createConnection();
    
            //3.打开连接
            connection.start();
    
            //4.创建session
            /** * 参数一：是否开启事务操作 * 参数二：消息确认机制 */
            Session session = connection.createSession( false, Session.AUTO_ACKNOWLEDGE );
    
            //5.创建目标地址（Queue:点对点消息，Topic：发布订阅消息）
            Topic topic = session.createTopic( "topic01" );
    
            //6.创建消息生产者
            MessageProducer producer = session.createProducer( topic );
    
            //7.创建消息
            //createTextMessage: 文本类型
            TextMessage textMessage = session.createTextMessage( "test message--topic" );
    
            //8.发送消息
            producer.send( textMessage );
    
            System.out.println( "消息发送完成" );
    
            //9.释放资源
            session.close();
            connection.close();
        }
    
    }

jms\_consumer

/** * 演示点对点模式- 消息消费者（第一种方案）阻塞式的 * * @author zrj * @date 2021/1/9 * @since V1.0 **/
    public class PTP_Consumer1 { 
    
        public static void main(String[] args) throws JMSException { 
            //1.创建连接工厂
            ConnectionFactory factory = new ActiveMQConnectionFactory( "tcp://127.0.0.1:61616" );
    
            //2.创建连接
            Connection connection = factory.createConnection();
    
            //3.打开连接
            connection.start();
    
            //4.创建session
            Session session = connection.createSession( false, Session.AUTO_ACKNOWLEDGE );
    
            //5.指定目标地址
            Queue queue = session.createQueue( "queue01" );
    
            //6.创建消息的消费者
            MessageConsumer consumer = session.createConsumer( queue );
    
            //7.接收消息
            while (true) { 
                Message message = consumer.receive();
    
                //如果已经没有消息了，结束啦
                if (message == null) { 
                    break;
                }
    
                //如果还有消息，判断什么类型的消息
                if (message instanceof TextMessage) { 
                    TextMessage textMessage = (TextMessage) message;
    
                    System.out.println( "接收的消息：" + textMessage.getText() );
                }
            }
        }
    
    }

/** * 演示点对点模式- 消息消费者（第二种方案）监听的方式 -- 更加推荐 * * @author zrj * @date 2021/1/9 * @since V1.0 **/
    public class PTP_Consumer2 { 
    
        /** * 注意：在监听器的模式下千万不要关闭连接，一旦关闭，消息无法接收 * * @param args * @throws JMSException */
        public static void main(String[] args) throws JMSException { 
            //1.创建连接工厂
            ConnectionFactory factory = new ActiveMQConnectionFactory( "tcp://127.0.0.1:61616" );
    
            //2.创建连接
            Connection connection = factory.createConnection();
    
            //3.打开连接
            connection.start();
    
            //4.创建session
            Session session = connection.createSession( false, Session.AUTO_ACKNOWLEDGE );
    
            //5.指定目标地址
            Queue queue = session.createQueue( "queue01" );
    
            //6.创建消息的消费者
            MessageConsumer consumer = session.createConsumer( queue );
    
            //7.设置消息监听器来接收消息
            consumer.setMessageListener( new MessageListener() { 
                //处理消息
                @Override
                public void onMessage(Message message) { 
                    if (message instanceof TextMessage) { 
                        TextMessage textMessage = (TextMessage) message;
    
                        try { 
                            System.out.println( "接收的消息（2）：" + textMessage.getText() );
                        } catch (JMSException e) { 
                            e.printStackTrace();
                        }
                    }
                }
            } );
    
        }
    }

/** * 演示发布订阅模式- 消息消费者 * * @author zrj * @date 2021/1/9 * @since V1.0 **/
    public class PS_Consumer { 
    
        /** * 注意：在监听器的模式下千万不要关闭连接，一旦关闭，消息无法接收 * 发布订阅模式时，需要先开启消费者监听，否则开启消费者之前的消息无法消费 * * @param args * @throws JMSException */
        public static void main(String[] args) throws JMSException { 
            //1.创建连接工厂
            ConnectionFactory factory = new ActiveMQConnectionFactory( "tcp://127.0.0.1:61616" );
    
            //2.创建连接
            Connection connection = factory.createConnection();
    
            //3.打开连接
            connection.start();
    
            //4.创建session
            Session session = connection.createSession( false, Session.AUTO_ACKNOWLEDGE );
    
            //5.指定目标地址
            Topic topic = session.createTopic( "topic01" );
    
            //6.创建消息的消费者
            MessageConsumer consumer = session.createConsumer( topic );
    
            //7.设置消息监听器来接收消息
            consumer.setMessageListener( new MessageListener() { 
                //处理消息
                @Override
                public void onMessage(Message message) { 
                    if (message instanceof TextMessage) { 
                        TextMessage textMessage = (TextMessage) message;
    
                        try { 
                            System.out.println( "接收的消息--topic：" + textMessage.getText() );
                        } catch (JMSException e) { 
                            e.printStackTrace();
                        }
                    }
                }
            } );
    
        }
    
    }

## 七、Spring整合ActiveMQ案例 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 13]spring\_producer

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>com.activemq</groupId>
        <artifactId>spring_producer</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <dependencies>
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-all</artifactId>
                <version>5.11.2</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-core</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-web</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-oxm</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-tx</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-jdbc</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-webmvc</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-aop</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-context-support</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-test</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-jms</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>javax.jms</groupId>
                <artifactId>javax.jms-api</artifactId>
                <version>2.0.1</version>
            </dependency>
            <dependency>
                <groupId>org.apache.xbean</groupId>
                <artifactId>xbean-spring</artifactId>
                <version>3.7</version>
            </dependency>
    
            <dependency>
                <groupId>junit</groupId>
                <artifactId>junit</artifactId>
                <version>4.12</version>
            </dependency>
        </dependencies>
    </project>

<?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:amp="http://activemq.apache.org/schema/core" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
    
        
        <amp:connectionFactory id="connetionFactory" brokerURL="tcp://127.0.0.1:61616" userName="admin" password="admin" />
    
        
        <bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
            
            <property name="targetConnectionFactory" ref="connetionFactory"/>
            
            <property name="sessionCacheSize" value="5"/>
        </bean>
    
        
        <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
            
            <property name="connectionFactory" ref="cachingConnectionFactory"/>
            
            <property name="pubSubDomain" value="false"/>
        </bean>
    
        
        <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">
            
            <property name="connectionFactory" ref="cachingConnectionFactory"/>
            
            <property name="pubSubDomain" value="true"/>
        </bean>
    </beans>

/** * 演示Spring与ActiveMQ整合 * @author zrj * @date 2021/1/9 * @since V1.0 **/
    @RunWith(SpringJUnit4ClassRunner.class) // junit与spring整合
    @ContextConfiguration("classpath:applicationContext-producer.xml") // 加载spring配置文件
    public class SpringProducer { 
        //点对点模式
        @Autowired
        @Qualifier("jmsQueueTemplate")
        private JmsTemplate jmsQueueTemplate;
    
        //发布订阅模式
        @Autowired
        @Qualifier("jmsTopicTemplate")
        private JmsTemplate jmsTopicTemplate;
    
    
        /** * 点对点发送 */
        @Test
        public void ptpSender(){ 
            /** * 参数一：指定队列的名称 * 参数二：MessageCreator接口，我们需要提供该接口的匿名内部实现 */
            jmsQueueTemplate.send("spring_queue", new MessageCreator() { 
                //我们只需要返回发送的消息内容即可
                @Override
                public Message createMessage(Session session) throws JMSException { 
                    //创建文本消息
                    TextMessage textMessage = session.createTextMessage("spring test message--queue");
                    return textMessage;
                }
            });
            System.out.println("消息发送已完成");
        }
    
        /** * 发布订阅发送 */
        @Test
        public void psSender(){ 
            jmsTopicTemplate.send("spring_topic", new MessageCreator() { 
                @Override
                public Message createMessage(Session session) throws JMSException { 
                    //创建文本消息
                    TextMessage textMessage = session.createTextMessage("spring test message--topic");
                    return textMessage;
                }
            });
            System.out.println("消息发送已完成");
        }
    }

spring\_consumer

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>com.activemq</groupId>
        <artifactId>spring_consumer</artifactId>
        <version>1.0-SNAPSHOT</version>
        
        <dependencies>
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-all</artifactId>
                <version>5.11.2</version>
            </dependency>
    
    
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-core</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-web</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-oxm</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-tx</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-jdbc</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-webmvc</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-aop</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-context-support</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-test</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-jms</artifactId>
                <version>5.0.2.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>javax.jms</groupId>
                <artifactId>javax.jms-api</artifactId>
                <version>2.0.1</version>
            </dependency>
            <dependency>
                <groupId>org.apache.xbean</groupId>
                <artifactId>xbean-spring</artifactId>
                <version>3.7</version>
            </dependency>
    
            <dependency>
                <groupId>junit</groupId>
                <artifactId>junit</artifactId>
                <version>4.12</version>
            </dependency>
        </dependencies>
    </project>

<?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:amp="http://activemq.apache.org/schema/core" xmlns:context="http://www.springframework.org/schema/context" xmlns:jms="http://www.springframework.org/schema/jms" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms.xsd">
    
        
        <amp:connectionFactory id="connectionFactory" brokerURL="tcp://127.0.0.1:61616" userName="admin" password="admin" />
    
    
        
        <bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
            <property name="targetConnectionFactory" ref="connectionFactory"/>
            <property name="sessionCacheSize" value="5"/>
        </bean>
    
    
        
        <context:component-scan base-package="com.activemq.listener"/>
    
        
        
        <jms:listener-container connection-factory="cachingConnectionFactory" destination-type="queue">
            <jms:listener destination="spring_queue" ref="queueListener"/>
        </jms:listener-container>
    
    
        
        <jms:listener-container connection-factory="cachingConnectionFactory" destination-type="topic">
            <jms:listener destination="spring_topic" ref="topicListener"/>
        </jms:listener-container>
    
    </beans>

/** * 点对点 * * @author zrj * @date 2021/1/9 * @since V1.0 **/
    @Component // 放入SpringIIOC容器，名称queueListener
    public class QueueListener implements MessageListener { 
    
        /** * 用于接收消息 * * @param message */
        @Override
        public void onMessage(Message message) { 
            if (message instanceof TextMessage) { 
    
                TextMessage textMessage = (TextMessage) message;
                try { 
                    System.out.println( "queue接口消息：" + textMessage.getText() );
                } catch (JMSException e) { 
                    e.printStackTrace();
                }
    
            }
    
        }
    }

/** * 发布订阅 * * @author zrj * @date 2021/1/9 * @since V1.0 **/
    @Component // 放入SpringIIOC容器，名称queueListener
    public class TopicListener implements MessageListener { 
    
        /** * 用于接收消息 * * @param message */
        @Override
        public void onMessage(Message message) { 
            if (message instanceof TextMessage) { 
    
                TextMessage textMessage = (TextMessage) message;
                try { 
                    System.out.println( "topic接口消息：" + textMessage.getText() );
                } catch (JMSException e) { 
                    e.printStackTrace();
                }
    
            }
    
        }
    }

/** * 用于启动消费方监听 * * @author zrj * @date 2021/1/9 * @since V1.0 **/
    public class SpringConsumer { 
    
        /** * 同时启动两个消费者 * spring_queue，spring_topic * * @param args * @throws IOException */
        public static void main(String[] args) throws IOException { 
            //1.加载spring配置
            ClassPathXmlApplicationContext cxt = new ClassPathXmlApplicationContext( "classpath:applicationContext-consumer.xml" );
            //2.启动
            cxt.start();
    
            //3.阻塞方法,让程序一直处于等待状态
            System.in.read();
    
        }
    }

## 八、SpringBoot整合ActiveMQ案例 ##

涉及情况比较多单独列一篇  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 14]springboot\_producer

server:
      port: 9001 #端口
    spring:
      application:
        name: activemq-producer # 服务名称
    
    # springboot与activemq整合配置
      activemq:
        broker-url: tcp://127.0.0.1:61616 # 连接地址
        user: admin # activemq用户名
        password: admin  # activemq密码
    
    # 指定发送模式 （点对点 false ， 发布订阅 true）
      jms:
        pub-sub-domain: true
    
    # 自己定义目标名称（队列或主题）
    activemq:
      name: springboot_topic

/** * 演示SpringBoot与ActiveMQ整合- 消息生产者 * * @author zrj * @date 2021/1/9 * @since V1.0 **/
    @RunWith(SpringJUnit4ClassRunner.class)
    @SpringBootTest(classes = ProducerApplication.class)
    public class SpringBootProducer { 
    
        //JmsMessagingTemplate: 用于工具类发送消息
        @Autowired
        private JmsMessagingTemplate jmsMessagingTemplate;
    
        @Value("${activemq.name}")
        private String name;
    
        @Test
        public void ptpSender() { 
            /** * 参数一：队列的名称或主题名称 * 参数二：消息内容 */
            jmsMessagingTemplate.convertAndSend( name, "spring boot message--topic" );
    
        }
    }

springboot\_consumer

server:
      port: 9002 #端口
    spring:
      application:
        name: activemq-consumer # 服务名称
    
    # springboot与activemq整合配置
      activemq:
        broker-url: tcp://127.0.0.1:61616 # 连接地址
        user: admin # activemq用户名
        password: admin  # activemq密码
    
    # 指定发送模式 （点对点 false ， 发布订阅 true）
      jms:
        pub-sub-domain: true
    
    activemq:
      name: springboot_topic

/** * 用于监听消息类（既可以用于队列的监听，也可以用于主题监听） * * @author zrj * @date 2021/1/9 * @since V1.0 **/
    @Component // 放入IOC容器
    public class MsgListener { 
    
        /** * 用于接收消息的方法 * destination: 队列的名称或主题的名称 */
        @JmsListener(destination = "${activemq.name}")
        public void receiveMessage(Message message) { 
            if (message instanceof TextMessage) { 
                TextMessage textMessage = (TextMessage) message;
                try { 
                    System.out.println( "接收消息：" + textMessage.getText() );
                } catch (JMSException e) { 
                    e.printStackTrace();
                }
    
            }
        }
    }

/** * 消息消费者启动类 * * @author zrj * @date 2021/1/9 * @since V1.0 **/
    @SpringBootApplication
    public class ConsumerApplication { 
        public static void main(String[] args) { 
            SpringApplication.run( ConsumerApplication.class, args );
        }
    }

[MQ]: https://blog.csdn.net/m0_37583655/article/details/112255224
[ActiveMQ]: https://blog.csdn.net/aoe15yu/article/details/104301780
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20210109214958877.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 1]: /images/20221119/8d7be435b10d4388a928fdda2cb90c88.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20210109215435943.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 3]: /images/20221119/3e128884e9db41d0837e45c4c36817e3.png
[http_activemq.apache.org]: http://activemq.apache.org/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 4]: /images/20221119/40ae24059895434bad0dd1112284f6b2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 5]: /images/20221119/4c302dae7a3541ed83df86ca51592e62.png
[http_localhost_8161_admin_queues.jsp]: http://localhost:8161/admin/queues.jsp
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20210109222718756.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 7]: /images/20221119/80a2077d447545e2ad66b27f9aeb4591.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/2021010922390973.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 9]: https://img-blog.csdnimg.cn/20210109224115806.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 10]: https://img-blog.csdnimg.cn/2021010922482357.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 11]: https://img-blog.csdnimg.cn/20210109225302149.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 12]: https://img-blog.csdnimg.cn/20210109225614896.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 13]: https://img-blog.csdnimg.cn/20210109230338826.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 14]: https://img-blog.csdnimg.cn/20210109231400484.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1,size_16,color_FFFFFF,t_70