学习ActiveMQ?这篇不错哦,错过就找不到了

川长思鸟来 2023-07-23 12:57 1阅读 0赞

**前言: 认真学完,相信你会有收获(＾－＾)V**

### 文章目录 ###

*  1. 什么是ActiveMQ?
 *  2.ActiveMQ工作原理: 注册发送短信验证举例
 *  3.ActiveMQ应用场景分析:
 *   *  3.1 当系统使用短信平台的时候
     *  3.2 当系统使用搜索平台、缓存平台的时候
 *  3.3. 总结
 *  4.安装
 *   *  4.1 下载
     *  4.2 运行
     *   *  4.2.1 解压并启动
         *  4.2.2 登录
     *  4.3 小小介绍一下,后面详细介绍
 *  5.使用Java操作ActiveMQ
 *   *  5.1 创建普通maven项目
     *  5.1.1 添加pom依赖
     *  5.1.2 编写生产者
     *   *  5.1.2.1 创建包com.manlu.queue , 创建类ActiveMQProducter
         *  5.1.2.2 查看控制台,消息发送成功
         *  5.1.2.3 查看消息是否发送成功(访问安装时登录的网址)
     *  5.1.3 编写消费者
     *   *  5.1.3.1 创建类ActiveMQConsumer,使用MessageConsumer完成消费
         *  5.1.3.2 查看控制台,可以发现消息已经被消费
         *  5.1.3.3 查看页面效果(访问安装时登录的网址)
     *  5.1.4 监听器消费消息
     *  5.2 多消费模式-queue
     *  5.3 Topic
     *  5.4 多消费模式-topic
     *   *  5.4.1 queue和topic模式的比较
 *  6.消息队列应用场景
 *   *  6.1 异步处理
     *  6.2 应用解耦
     *  6.3 流量削峰
     *  6.4 日志处理
     *  6.5 消息通讯
 *  7. JMS消息服务
 *   *  7.1 消息模型
     *  7.2 P2P模式-队列模式
     *  7.3 Pub/Sub模式-广播/主题模式
 *  8.SpringBoot整合ActiveMQ
 *   *  8.1 创建工程maven子模块
     *   *  8.1.1 创建mq\_springBoot子模块
         *  8.1.2 添加相关依赖
         *  8.1.3 创建程序入口
         *  8.1.4 创建application.yml文件
         *  8.1.5 创建config
     *  8.2 编写消费者和生产者
     *   *  8.2.1 编写生产者QueueProducer
         *  8.2.2 编写消费者QueueConsumer
         *  8.2.3 启动
         *  注意: 启动前需要先运行你的ActiveMQ
     *  如果出现问题, 评论即可, 我看到会尝试解决

# 1. 什么是ActiveMQ? #

ActiveMQ是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ是一个完全支持JMS1.1和J2EE1.4规范的
    JMS Provide实现，尽管JMS规范出台是很久的事情了，但是JMS在当今的J2EE应用中间任然扮演着特殊的地位.

# 2.ActiveMQ工作原理: 注册发送短信验证举例 #

![format_png][]

1. 解决服务之间代码耦合
    2. 使用消息队列, 增加系统并发处理量

![format_png 1][]

# 3.ActiveMQ应用场景分析: #

## 3.1 当系统使用短信平台的时候 ##

用户注册,重点使用用户信息数据库保存，而发短信、发邮件，增加业务处理复杂度，这时候用
    MQ，将发短信、发邮箱、通知MQ，由另外服务平台完成。解决了代码的耦合问题。

## 3.2 当系统使用搜索平台、缓存平台的时候 ##

查询数据，建立缓存、索引 ，当再次查询相同数据的时候，不从数据库查询，从缓存或者索引
    库查询
       当增加、修改、删除数据时，发送消息给MQ， 缓存平台、索引平台 从MQ获取到这个信息，更新
    缓存或者索引

# 3.3. 总结 #

使用MQ作为系统间数据调用的中转站。

![format_png 2][]

# 4.安装 #

官网：http://activemq.apache.org/
    
    百度网盘地址:
    链接：https://pan.baidu.com/s/1A55X-vchhsaxaoaIXDUw8g 
    提取码：qby6
    复制这段内容后打开百度网盘手机App，操作更方便哦

## 4.1 下载 ##

进入官网，向下滑动到这个位置

![format_png 3][]

点击下载最新windows版

![format_png 4][]

## 4.2 运行 ##

### 4.2.1 解压并启动 ###

将下载好的压缩文件，根据自己的需求放到指定目录，然后解压，
    解压后到apache-activemq-5.15.12\bin\win64目录 启动activemq.bat 文件

![format_png 5][]  
![format_png 6][]

### 4.2.2 登录 ###

访问:http://localhost:8161
    点击 Manage ActiveMQ broker,点击去下面标识的即可看到

![format_png 7][]

账号和密码都是admin
    http://localhost:8161/admin/

![format_png 8][]

登录成功,看到如下页面

![format_png 9][]

## 4.3 小小介绍一下,后面详细介绍 ##

ActiveMQ使用的是标准生产者和消费者模型

1. Queue队列(P2P消息模型)，生产者生产了一个消息，只能由一个消费者进行消费。如给微信好友发消息。
    2. Topic主题/广播(Pub/Sub消息模型)，生产者生产了一个消息，可以由多个消费者进行消费。如微信公众号给粉丝发消息。

JMS和ActiveMQ的对应

<table> 
 <thead> 
  <tr> 
   <th>JMS消息模型</th> 
   <th>P2P模式</th> 
   <th>pub/sub模式</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ActiveMQ消息</td> 
   <td>Queue队列</td> 
   <td>Topic队列</td> 
  </tr> 
  <tr> 
   <td>特点</td> 
   <td>一对一,一个人发送,只允许一个人接收</td> 
   <td>一对多,一个人发送, 允许多个人接收</td> 
  </tr> 
 </tbody> 
</table>

发送的人: 生产者
    接收的人: 消费者

# 5.使用Java操作ActiveMQ #

## 5.1 创建普通maven项目 ##

简单的截几张创建时的图

![format_png 10][]

![format_png 11][]

![format_png 12][]

## 5.1.1 添加pom依赖 ##

<dependencies>
      <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-all</artifactId>
        <version>5.15.0</version>
      </dependency>
    </dependencies>

如果自己仓库没有依赖, 导入后会自己下载, 稍等即可

![format_png 13][]

## 5.1.2 编写生产者 ##

使用JMS原生API编写测试类, 小消息中间件写入消息的开发步骤如下:

1. 创建连接工厂
    2. 从连接工厂中获取连接
    3. 启动连接
    4. 获取会话
    5. 创建Queue队列
    6. 创建生产者
    7. 创建消息
    8. 发送消息
    9. 提交请求
    10. 关闭各种资源

### 5.1.2.1 创建包com.manlu.queue , 创建类ActiveMQProducter ###

![format_png 14][]

package com.manlu.queue;
    import org.apache.activemq.ActiveMQConnectionFactory;
    import javax.jms.*;
    
    /** * @author 漫路h * 生成者发送消息 */
    public class ActiveMQProducter { 
        public static void main(String[] args) throws Exception { 
            // 连接工厂
            // 使用默认用户名、密码、路径
            // 因为: 底层实现: final String defaultURL = "tcp://"+DEFAULT_BROKER_HOST+ ":" + DEFAULT_BROKER_PORT;
            // 所以: 路径 tcp://host:61616
            // 1. 创建连接工厂
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
            // 2. 创建连接
            Connection connection = connectionFactory.createConnection();
            // 3. 打开连接
            connection.start();
            // 4. 创建会话
            //第一个参数: 是否开启事务
            //第二个参数: 消息是否自动确认
            Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            //创建队列
            Queue queue = session.createQueue("hello");
            // 5. 创建生产者
            MessageProducer producer = session.createProducer(queue);
            // 6. 创建消息
            TextMessage message = session.createTextMessage("hi i am manlu");
            // 7. 发送消息
            producer.send(message);
            // 8. 关闭资源
            session.commit();
            producer.close();
            session.close();
            connection.close();
            System.out.println("消息生成成功");
        }
    }

### 5.1.2.2 查看控制台,消息发送成功 ###

![format_png 15][]

### 5.1.2.3 查看消息是否发送成功(访问安装时登录的网址) ###

默认tcp连接activeMQ端口 61616 ！！

![format_png 16][]

## 5.1.3 编写消费者 ##

使用JMS原生API编写测试类，向消息中间件消费消息的开发步骤：

1. 创建链接工厂
    2. 创建链接
    3. 启动链接
    4. 获取会话
    5. 创建队列
    6. 创建消费者
    7. 消费消息
    8. 提交
    9. 关闭资源

### 5.1.3.1 创建类ActiveMQConsumer,使用MessageConsumer完成消费 ###

package com.manlu.queue;
    import org.apache.activemq.ActiveMQConnectionFactory;
    import javax.jms.*;
    
    /** * @author manlu * 使用MessageConsumer完成消费 */
    public class ActiveMQConsumer { 
        public static void main(String[] args) throws Exception { 
            // 1.创建连接工厂
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
            // 2.创建连接
            Connection connection = connectionFactory.createConnection();
            // 3.开启连接
            connection.start();
            // 4.创建会话
            /* * 第一个参数,是否使用事务 * 如果设置true,操作消息队列后, 必须使用 session.commit(); * 如果设置false, 操作消息队列后, 不使用 session.commit(); */
            Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            // 5.创建队列
            Queue queue = session.createQueue("hello");
            // 6.创建消费者
            MessageConsumer consumer = session.createConsumer(queue);
            while (true){ 
                //失效时间, 如果10秒内没有收到新的消息,说明没有消息存在,此时可以退出当前循环
                TextMessage message = (TextMessage) consumer.receive(10000);
                if (message!=null) { 
                    System.out.println(message.getText());
                }else{ 
                    break;
                }
            }
            // 7.关闭连接
            session.commit();
            session.close();
            connection.close();
            System.out.println("消费结束");
        }
    }

### 5.1.3.2 查看控制台,可以发现消息已经被消费 ###

![format_png 17][]

### 5.1.3.3 查看页面效果(访问安装时登录的网址) ###

*  消费前

![format_png 18][]

*  消费后

![format_png 19][]

## 5.1.4 监听器消费消息 ##

package com.manlu.queue;
    import org.apache.activemq.ActiveMQConnectionFactory;
    import javax.jms.*;
    
    /** * @author 漫路 * 监听器消费消息 * 重复测试生成和消费的过程。实现一边生成，一边消费的系统 */
    public class ActiveMQMonitorConsumer { 
        public static void main(String[] args) throws Exception { 
            // 1.创建连接工厂
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
            // 2.创建连接
            Connection connection = connectionFactory.createConnection();
            // 3.开启连接
            connection.start();
            // 4.创建会话
            /* * 第一个参数,是否使用事务 * 如果设置true,操作消息队列后, 必须使用 session.commit(); * 如果设置false, 操作消息队列后, 不使用 session.commit(); */
            Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            // 5.创建队列
            Queue queue = session.createQueue("hello");
            // 6.创建消费者
            MessageConsumer messageConsumer = session.createConsumer(queue);
            messageConsumer.setMessageListener(new MessageListener() { 
                //每次接受消息,自动调用 onMessage
                @Override
                public void onMessage(Message message) { 
                    TextMessage textMessage = (TextMessage) message;
                    try { 
                        System.out.println(textMessage.getText());
                    } catch (JMSException e) { 
                        e.printStackTrace();
                    }
                }
            });
            //我们不让程序结束,因为如果结束, 监听就结束了
            while (true){ 
                //目的: 不让程序死掉
            }
        }
    }

## 5.2 多消费模式-queue ##

P2P消息模型中的多消费者模式,得出结论如下:
    	- 一个消息只能被一个消费者消费, 不可重复消费
    	- 多个消费者均分消息(负载均衡策略)
    	- 当消费者在消费某个消息的时候, mq一定要等到它的成功回执, 才会分发下一个消息

![format_png 20][]  
注意: 测试的时候一定要先启动消费者,再启动生产者

## 5.3 Topic ##

Topic：主题模式、广播模式、pus/sub模式

// 1. 创建连接工厂
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
            // 2. 创建连接
            Connection connection = connectionFactory.createConnection();
            // 3. 打开连接
            connection.start();
            // 4. 创建会话
            //第一个参数: 是否开启事务
            //第二个参数: 消息是否自动确认
            Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
    	// 5.创建Topic主题模式
    	Topic topic = session.createTopic("manlu123");
    	// 6.创建生产者
    	MessageProducer producer = session.createProducer(topic);

## 5.4 多消费模式-topic ##

### 5.4.1 queue和topic模式的比较 ###

相同点:
      1. 都只有一个生产者
      2. 都可以有多个消费者
    不同点:
      1. queue队列模式, 一个消息只能别一个消费者消费, 不能重复消费
      	当消费者消费某个消息的时候,一定要得到这个消息被成功消费的回执,才会分发下一个消息
    	queue入队之后,无论等待多久, 消息都会一直等待消费者来处理
      2. topic广播模式, 一个消息可以被多个消费者消费
      	这个消息无论被成功消费与否,都无所谓
    	topic要求时间要一致, 我正好发, 你正好收

# 6.消息队列应用场景 #

异步处理、应用解耦、流量削峰和信息通讯四个场景

## 6.1 异步处理 ##

*  场景说明：用户注册后，需要发注册邮件和注册短信。传统的做法有两种: 1.串行方式；2.并行方式
    
     *  1.串行方式：将注册信息写入 数据库 成功后，发送注册邮件，再发送注册短信。以上三个任务全部完成后，返回给客户端  
        ![format_png 21][]
     *  2.并行方式: 将注册信息写入数据库成功后，发送注册邮件的同时，发送注册短信。以上三个任务完成后，返回给客户端。与串行的差别是，并行的方式可以提高处理的时间  
        ![format_png 22][]

假设三个业务节点每个使用50毫秒钟，不考虑网络等其他开销，则串行方式的时间是150毫秒，
    并行的时间可能是100毫秒。
       因为CPU在单位时间内处理的请求数是一定的，假设CPU在1秒内吞吐量是100次。则串行方式1秒
    内CPU可处理的请求量是7次（1000/150）。并行方式处理的请求量是10次（1000/100）
       小结：如以上案例描述，传统的方式系统的性能（并发量，吞吐量，响应时间）会有瓶颈。如何解
    决这个问题呢？
    引入消息队列，将不是必须的业务逻辑，进行异步处理。改造后的架构如下：

![format_png 23][]

## 6.2 应用解耦 ##

*  场景说明：用户下单后，订单系统需要通知库存系统。传统的做法是，订单系统调用库存系统的接口。如下图

![format_png 24][]

*  传统模式的缺点:
    
     *  1.  加入库存系统无法访问，则订单减库存将失败，从而导致订单失败
     *  1.  订单系统与库存系统耦合
 *  以上问题如何解决？引入应用消息队列后的方案，如下图:

![format_png 25][]

订单系统：用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单。下单成功
    库存系统：订阅下单的消息，采用pub/sub(发布/订阅)的方式，获取下单信息，库存系统根据
    	 下单信息，进行库存操作
    假如：在下单时库存系统不能正常使用。也不影响正常下单，因为下单后，订单系统写入消息
         队列就不再关心其他的后续操作了。实现订单系统与库存系统的应用解耦

## 6.3 流量削峰 ##

*  应用场景：秒杀活动，一般会因为流量过大，导致流量暴增，应用挂掉。为解决这个问题，一般需要在应用前端加入消息队列。
    
     *  1.可以控制活动人数
     *  2.可以缓解短时间内高流量压垮应用
     *  3.用户请求，服务器接收后，首先写入消息队列。假如消息队列长度超过最大数量，则直接抛弃用户请求或跳转到错误页面
     *  4.秒杀业务根据消息队列中的请求信息，再做后续处理

![format_png 26][]

## 6.4 日志处理 ##

日志处理是指将消息队列用在日志处理中，比如Kafka的应用，解决大量日志传输的问题。架构简化如下  
![format_png 27][]

*  日志采集客户端，负责日志数据采集，定时写入到Kafka队列
 *  Kafka消息队列，负责日志数据的接收，存储和转发
 *  日志处理应用：订阅并消费kafka队列中的日志数据

![format_png 28][]

1）Kafka：接收用户日志的消息队列
     2）Logstash：做日志解析，统一成JSON输出给Elasticsearch
     3）Elasticsearch：实时日志分析服务的核心技术，一个schemaless，实时的数据存储服务，通过index组织数据，
     		   兼具强大的搜索和统计功能
     4）Kibana：基于Elasticsearch的数据可视化组件，超强的数据可视化能力是众多公司选择ELKstack的重要原因

## 6.5 消息通讯 ##

消息通讯是指，消息队列一般都内置了高效的通信机制，因此也可以用在纯的消息通讯。比如实现点对点消息队列，或者聊天室等

*  点对点通讯:

客户端A和客户端B使用同一队列，进行消息通讯。

![format_png 29][]

*  聊天室通讯：

客户端A，客户端B，客户端N订阅同一主题，进行消息发布和接收。实现类似聊天室效果。以上实际是消息队列的两种消息模式，点对点或发布订阅模式。模型为示意图，供参考。

![format_png 30][]

# 7. JMS消息服务 #

消息队列的JavaEE规范JMS。JMS(Java Message Service即Java消息服务) API是一个信息服务的标准/规范,允许应用程序组件基于JavaEE平台创建、发送、接收和读取消息。它使分布式通信耦合度更低，消息服务更加可靠以及异步性。

## 7.1 消息模型 ##

在JMS标准中，有两种消息模型P2P（Point to Point），Publish/Subscribe（Pub/Sub）。

## 7.2 P2P模式-队列模式 ##

![format_png 31][]

P2P模式包含三个角色：消息队列（Queue），发送者（Sender）、接收者（Receiver）。每个消息都被发送到一个特定的队列，接收者从队列中获取消息。队列保留着消息，知道他们被消费或超时。

*  P2P的特点
    
     *  每个消息只能被一个消费者（Consumer）消费（即一旦被消费，信息就不再存在于消息队列中）
     *  发送者和接收者之间在时间上没有依赖性，也就是说当发送者发送了消息之后，不管接收者有没有正在运行，它不会影响到消息被发送到队列
     *  接收者在成功接收到消息之后需要向队列应答成功

如果希望发送的每个消息都会被成功处理的话，那么需要P2P模式。

## 7.3 Pub/Sub模式-广播/主题模式 ##

![format_png 32][]

包含三个角色：主题（Topic）、发布者（Publisher）、订阅者（Subscriber）多个发布者将消息发送到Topic，系统将这些消息传递给多个订阅者。

*  Pub/Sub的特点
    
     *  每个消息可以有多个消费者
     *  发布者和订阅者之间有时间上的依赖性。针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能被消费发布者的消息
     *  为了消费消息，订阅者必须保持运行的状态

为了缓和这样严格的时间相关性，JMS允许订阅者创建一个可持久化的订阅。这样，即使订阅者没有被激活（运行），它也能接收到发布者的信息。

如果希望发送的消息可以被多个消费者处理的话，那么可以采用Pub/Sub模式

*  消息消费方式

在JMS中，消息的产生和消费都是异步的。对于消费来说，JMS的消费者可以通过两种方式来消费消息。

1）同步：
    订阅者或接收者通过receive方法来接收消息，receive在接收消息之前（或超时之前）将一直堵塞；
    2）异步
    订阅者或消费者可以注册为一个消息监听器。当消息到达之后，系统自动调用监听器的onMessage方法。

# 8.SpringBoot整合ActiveMQ #

在SpringBoot中集成ActiveMQ相对还是比较简单的，不需要安装什么服务，
    默认使用内存中的ActiveMQ，配合外置ActiveMQ Server会更好.

## 8.1 创建工程maven子模块 ##

### 8.1.1 创建mq\_springBoot子模块 ###

![format_png 33][]

### 8.1.2 添加相关依赖 ###

<dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-activemq</artifactId>
      <version>2.0.5.RELEASE</version>
    </dependency>

### 8.1.3 创建程序入口 ###

![format_png 34][]

package com.manlu;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    /** * @author 漫路 */
    @SpringBootApplication
    public class MqSbApplication { 
        public static void main(String[] args) { 
            SpringApplication.run(MqSbApplication.class,args);
        }
    }

### 8.1.4 创建application.yml文件 ###

![format_png 35][]

spring:
      activemq:
        broker-url: tcp://127.0.0.1:61616 #MQ所在的服务器地址
        in-memory: true #是否使用内置的MQ。 true：使用； false：不使用；
        non-blocking-redelivery: false #是否在回滚消息之前停止消息传递。当启用此命令时，消息顺序不会被保留
        user: admin # 用户名
        password: admin # 密码

*  配置的所有具体意思如下(properties格式):

spring.activemq.broker-url=tcp://127.0.0.1:61616
    # 在考虑结束之前等待的时间
    #spring.activemq.close-timeout=15s
    # 默认代理URL是否应该在内存中。如果指定了显式代理，则忽略此值。
    spring.activemq.in-memory=true
    # 是否在回滚回滚消息之前停止消息传递。这意味着当启用此命令时，消息顺序不会被保留。
    spring.activemq.non-blocking-redelivery=false
    # 密码
    spring.activemq.password=admin
    # 等待消息发送响应的时间。设置为0等待永远。
    spring.activemq.user=admin
    # 是否信任所有包
    #spring.activemq.packages.trust-all=
    # 要信任的特定包的逗号分隔列表（当不信任所有包时）
    #spring.activemq.packages.trusted=
    # 当连接请求和池满时是否阻塞。设置false会抛“JMSException异常”。
    #spring.activemq.pool.block-if-full=true
    # 如果池仍然满，则在抛出异常前阻塞时间。
    #spring.activemq.pool.block-if-full-timeout=-1ms
    # 是否在启动时创建连接。可以在启动时用于加热池。
    #spring.activemq.pool.create-connection-on-startup=true
    # 是否用Pooledconnectionfactory代替普通的ConnectionFactory。
    #spring.activemq.pool.enabled=false
    # 连接过期超时。
    #spring.activemq.pool.expiry-timeout=0ms
    # 连接空闲超时
    #spring.activemq.pool.idle-timeout=30s
    # 连接池最大连接数
    #spring.activemq.pool.max-connections=1
    # 每个连接的有效会话的最大数目。
    #spring.activemq.pool.maximum-active-session-per-connection=500
    # 当有"JMSException"时尝试重新连接
    #spring.activemq.pool.reconnect-on-exception=true
    # 在空闲连接清除线程之间运行的时间。当为负数时，没有空闲连接驱逐线程运行。
    #spring.activemq.pool.time-between-expiration-check=-1ms
    # 是否只使用一个MessageProducer
    #spring.activemq.pool.use-anonymous-producers=true

### 8.1.5 创建config ###

![format_png 36][]

package com.manlu.config;
    
    import org.apache.activemq.command.ActiveMQQueue;
    import org.apache.activemq.command.ActiveMQTopic;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import javax.jms.Queue;
    import javax.jms.Topic;
    
    /** * @author 漫路 */
    @Configuration
    public class ActiveMQConfig { 
        @Bean
        public Queue queue(){ 
            return new ActiveMQQueue("manlu.queue");
        }
        
        @Bean
        public Topic topic(){ 
            return new ActiveMQTopic("manlu.topic");
        }
    }

## 8.2 编写消费者和生产者 ##

### 8.2.1 编写生产者QueueProducer ###

![format_png 37][]

package com.manlu.mq;
    
    import org.apache.activemq.command.ActiveMQMapMessage;
    import org.springframework.jms.core.JmsMessagingTemplate;
    import org.springframework.scheduling.annotation.EnableScheduling;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    import javax.jms.MapMessage;
    import javax.jms.Queue;
    
    /** * 消息的生产者 * * @author 漫路 */
    @Component
    @EnableScheduling
    public class QueueProducer { 
        /* * @Resource // 也可以注入JmsTemplate，JmsMessagingTemplate对JmsTemplate进行了封装 * private JmsMessagingTemplate jmsTemplate; // * 发送消息，destination是发送到的队列，message是待发送的消息 * * @Scheduled(fixedDelay=3000)//每3s执行1次 * public void sendMessage(Destination destination, final String message){ * jmsTemplate.convertAndSend(destination, message); * } */
        @Resource
        private JmsMessagingTemplate jmsMessagingTemplate;
    
        @Resource
        private Queue queue;
    
        @Scheduled(fixedDelay = 3000)//3秒执行1次
        public void send() { 
            try { 
                MapMessage mapMessage = new ActiveMQMapMessage();
                mapMessage.setString("info", "小老弟在敲代码");
                jmsMessagingTemplate.convertAndSend(queue, mapMessage);
            } catch (Exception e) { 
                e.printStackTrace();
            }
    
        }
    }

### 8.2.2 编写消费者QueueConsumer ###

![format_png 38][]

package com.manlu.mq;
    
    import org.springframework.jms.annotation.JmsListener;
    import org.springframework.stereotype.Component;
    
    import javax.jms.JMSException;
    import javax.jms.MapMessage;
    import javax.jms.Message;
    
    /** * 消息的消费者 * @author 漫路 */
    @Component
    public class QueueConsumer { 
        //使用JmsListener配置消费者监听的队列,其中Message是接收到的消息
        @JmsListener(destination = "manlu.queue")
        public void receiveQueue(Message message){ 
            try { 
                MapMessage mapMessage = (MapMessage) message;
                String info = mapMessage.getString("info");
                System.out.println(info);
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }

### 8.2.3 启动 ###

### 注意: 启动前需要先运行你的ActiveMQ ###

如何运行上面已经讲过,点看目录找运行即可

*  启动springboot

![format_png 39][]

*  访问activemq看一下

![format_png 40][]

## 如果出现问题, 评论即可, 我看到会尝试解决 ##

[format_png]: /images/20230528/faa7289f6c5e4880a67a37577ae2efba.png
[format_png 1]: /images/20230528/bf7aadd752e040ea87266c25d20445c7.png
[format_png 2]: /images/20230528/cc440935852b4733a49c472dd30c6d4f.png
[format_png 3]: /images/20230528/c4e9183a8d134b87b7f672fc1ee094bf.png
[format_png 4]: /images/20230528/0107471f6c41468d910a8e69e84d9c57.png
[format_png 5]: /images/20230528/f4edaf8ab8434e0ea4bbd522aa800eca.png
[format_png 6]: /images/20230528/687ca05b1af54ce184aae419d1483c51.png
[format_png 7]: /images/20230528/45bc4bc09ca1405f8502aa0024ccecba.png
[format_png 8]: /images/20230528/3e6ec6fa0e6b48eb972067beedcf01c6.png
[format_png 9]: /images/20230528/8aba4e77c41d47bbb1d5ef44394b954d.png
[format_png 10]: /images/20230528/a0e64db64c6145789de08ff6c728f136.png
[format_png 11]: /images/20230528/10192ce3b63e4871a4bebb8506576645.png
[format_png 12]: /images/20230528/aaa41b6fda8245eb8308d3085e268e56.png
[format_png 13]: /images/20230528/e4c3c23215f447cc923f74657db61b68.png
[format_png 14]: /images/20230528/c49a714dbd5c488baf9d876f29468189.png
[format_png 15]: /images/20230528/eaa606844a9044b5bbc2851aa5c1751b.png
[format_png 16]: /images/20230528/a386e72222b74c6f954b8ab258f03b99.png
[format_png 17]: /images/20230528/4515e078951346ed9da997a06ea6c255.png
[format_png 18]: /images/20230528/5fa16c5716f948688abfb12b7977159f.png
[format_png 19]: /images/20230528/4d209e44b0fe456cb6116d0c36ec6450.png
[format_png 20]: /images/20230528/2be0fa2397b4448c9db9e2e70d284b93.png
[format_png 21]: /images/20230528/babab4ea09844ba6948d18c4d1aae4b9.png
[format_png 22]: /images/20230528/7e3dbef085ff4f028635edd9a74649ba.png
[format_png 23]: /images/20230528/d633babcd76942d0b5d25baa017905cd.png
[format_png 24]: /images/20230528/e0a92c6bd27d495f8675bf2777ce0836.png
[format_png 25]: /images/20230528/d6e62265210641d098166de2ad9e8e22.png
[format_png 26]: /images/20230528/77b09685c0b84ef8bdbbd6e6c9d668d7.png
[format_png 27]: /images/20230528/4964783269b741e0b3ae8d7bd8d68a47.png
[format_png 28]: /images/20230528/a4090ea82a41466fbd72bf342ff2f9e2.png
[format_png 29]: /images/20230528/adc07705d1424240bc304a64fc0b4ce5.png
[format_png 30]: /images/20230528/5cdfcf943a5348e6b0475f1780b24739.png
[format_png 31]: /images/20230528/0fc75fe17c694689a4c7b6763b0129a1.png
[format_png 32]: /images/20230528/3e695cb057244a73b8ada8497a09df1d.png
[format_png 33]: /images/20230528/de7dace47ad44d3a8693603010ed5f26.png
[format_png 34]: /images/20230528/aecb3a96804b4ebba64fddbd64e692de.png
[format_png 35]: /images/20230528/746771806c474a018f7a2c2a19ffb0f3.png
[format_png 36]: /images/20230528/f1b8259f3c694c1a868aa6ea83d4a798.png
[format_png 37]: /images/20230528/b653c0cd8b094d939ad33a161ab0349c.png
[format_png 38]: /images/20230528/18e2ceffde9c4f19bd0172b5b055ae47.png
[format_png 39]: /images/20230528/5c5fb4f7904646c5b3a01cc5f05097db.png
[format_png 40]: /images/20230528/90e465a7007145a7ad7ff9b7c41f0d35.png