ActiveMQ学习总结

一、消息中间件的产生的背景

1.在网络通讯中，Http请求默认采用同步请求方式，基于请求与响应模式

2.在客户端与服务器进行通讯时，客户端调用服务端接口后，必须等待服务端完成处理后返回结果给客户端才能继续执行，这种情况属于同步调用方式。

3.如果服务器端发生网络延迟、不可达的情况，可能客户端也会受到影响。

二、什么是消息中间件

消息队列中间件是分布式系统中重要的组件，主要解决应用解耦，异步消息，流量削锋等问题，实现高性能，高可用，可伸缩和最终一致性架构。

目前使用较多的消息队列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ

实现方式:面向消息的中间件(MessageOrlented MiddlewareMOM)较好的解决了以上问题。发送者将消息发送给消息服务器，消息服务器将消感存放在若千队列中，

在合适的时候再将消息转发给接收者。这种模式下，发送和接收是异步的，发送者无需等待; 二者的生命周期未必相同: 发送消息的时候接收者不一定运行，

接收消息的时候发送者也不一定运行;一对多通信: 对于一个消息可以有多个接收者。

原理图

三、JMS介绍

1、什么是JMS

JMS是java的消息服务器，JMS的客户端之间可以通过JMS服务进行异步的消息传输,是一个Java平台中面向消息中间件的API

角色划分

   1.提供者： 实现JMS规范的消息中间件服务器 (存放消息容器)

  2.客户端：发送或接收消息的应用程序

  3.生产者/发布者： 创建并发送消息的客户端（向消息容器存放消息）

  4.消费者/订阅者：接收并处理消息的客户端

  5.消息：应用程序之间传递的数据内容

  6.消息模式：在客户端之间传递消息的方式，JMS中定义了主题和队列两种模式 点对点与发布订阅模式。

2、什么是消息模型

•  Point-to-Point(P2P) --- 点对点（生产者发送一条消息到queue，只有一个消费者能收到）
• Publish/Subscribe(Pub/Sub)---  发布订阅(发布者发送到topic的消息，只有订阅了topic的订阅者才会收到消息)

     即点对点和发布订阅模型

    P2P(点对点)

• p2p模型图

• 相关概念

1. 消息队列（Queue）
2. 发送者(Sender)
3. 接收者(Receiver)
4. 每个消息都被发送到一个特定的队列，接收者从队列中获取消息。队列保留着消息，直到他们被消费或超时。

•  p2p特点

1. 每个消息只有一个消费者（Consumer）(即一旦被消费，消息就不再在消息队列中)
2. 发送者和接收者之间在时间上没有依赖性，也就是说当发送者发送了消息之后，不管接收者有没有正在运行，它不会影响到消息被发送到队列
3. 接收者在成功接收消息之后需向队列应答成功，如果你希望发送的每个消息都应该被成功处理的话，那么你需要P2P模式。

•  应用场景

             A用户与B用户发送消息

     Pub/Sub (发布与订阅)        

• Pub/Sub模式图

 

• 相关概念

            1.主题（Topic）

            2.发布者（Publisher）

            3.订阅者（Subscriber） 
            4.客户端将消息发送到主题。多个发布者将消息发送到Topic,系统将这些消息传递给多个订阅者。

• Pub/Sub的特点 

            每个消息可以有多个消费者

                         发布者和订阅者之间有时间上的依赖性。针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能消费发布者的消息，而且为了消费消息，订阅者必须保持运行的状态。

                  为了缓和这样严格的时间相关性，JMS允许订阅者创建一个可持久化的订阅。这样，即使订阅者没有被激活（运行），它也能接收到发布者的消息。  如果你希望发送的消息可以不被做任何处理、

                   或者被一个消息者处理、或者可以被多个消费者处理的话，那么可以采用Pub/Sub模型

           消息的消费 
                 在JMS中，消息的产生和消息是异步的。对于消费来说，JMS的消息者可以通过两种方式来消费消息。 
         ○ 同步 
                订阅者或接收者调用receive方法来接收消息，receive方法在能够接收到消息之前（或超时之前）将一直阻塞 
          ○ 异步 
               订阅者或接收者可以注册为一个消息监听器。当消息到达之后，系统自动调用监听器的onMessage方法。

•  发布订阅与点对点通讯方式区别：

                点对点 只能保证一个消费者进行消费 一对一 发布订阅 只要集群服务订阅该主题都会受收到消息 一对多

 四、ActiveMQ

1、ActiveMQ的消息形式

    应用场景:

•  异步处理
•  实现系统之间的解耦
• 流量削锋
• 消息通讯

异步处理应用场景

异步处理 场景说明：用户注册后，需要发注册邮件和注册短信。传统的做法有两种 1.串行的方式2.并行方式

a、串行方式：将注册信息写入数据库成功后，

发送注册邮件，再发送注册短信。以上三个任务全部完成后，返回给客户端。

b.并行方式：将注册信息写入数据库成功后，发送注册邮件的同时，发送注册短信。以上三个任务完成后，返回给客户端。

与串行的差别是，并行的方式可以提高处理的时间

引入消息队列，将不是必须的业务逻辑，异步处理。改造后的架构如下：

 

按照以上约定，用户的响应时间相当于是注册信息写入数据库的时间，也就是50毫秒。注册邮件，发送短信写入消息队列后，直接返回，因此写入消息队列的速度很快，基本可以忽略，因此用户的响应时间可能是50毫秒。因此架构改变后，系统的吞吐量提高到每秒20 QPS。比串行提高了3倍，比并行提高了两倍。

解耦应用场景

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

传统模式的缺点：假如库存系统无法访问，则订单减库存将失败，从而导致订单失败，订单系统与库存系统耦合 耗时时间接口 统一采用MQ推送 不建议才同步方

如何解决以上问题呢？引入应用消息队列后的方案，如下图：

 

订单系统：用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功 库存系统：订阅下单的消息，

采用拉/推的方式，获取下单信息，库存系统根据下单信息，进行库存操作 假如：在下单时库存系统不能正常使用。也不影响正常下单，

因为下单后，订单系统写入消息队列就不再关心其他的后续操作了。实现订单系统与库存系统的应用解耦

流量削峰应用解耦

流量削锋也是消息队列中的常用场景，一般在秒杀或团抢活动中使用广泛。 应用场景：秒杀活动，一般会因为流量过大，导致流量暴增，应用挂掉。

为解决这个问题，一般需要在应用前端加入消息队列。

a、可以控制活动的人数

b、可以缓解短时间内高流量压垮应用

 

用户的请求，服务器接收后，首先写入消息队列。假如消息队列长度超过最大数量，则直接抛弃用户请求或跳转到错误页面。

秒杀业务根据消息队列中的请求信息，再做后续处理 秒杀如何实现核心Redis+MQ+服务保护机制（服务降级、隔离、熔断）+服务限流+图形验证+token

对于消息的传递有两种类型：

•  一种是点对点的，即一个生产者和一个消费者一一对应；
• 另一种是发布/订阅模式，即一个生产者产生消息并进行发送后，可以由多个消费者进行接收。

JMS定义了五种不同的消息正文格式，以及调用的消息类型，允许你发送并接收以一些不同形式的数据，提供现有消息格式的一些级别的兼容性。

　　· StreamMessage -- Java原始值的数据流

　　· MapMessage--一套名称-值对

　　· TextMessage--一个字符串对象(常用)

　　· ObjectMessage--一个序列化的 Java对象

　　· BytesMessage--一个字节的数据流

2、activeMQ window安装

进入http://activemq.apache.org/下载ActiveMQ

下载window或者Linux版本

ActiveMQ部署其实很简单，和所有Java一样，要跑java程序就必须先安装JDK并配置好环境变量，这个很简单。

然后解压下载的apache-activemq-5.10-20140603.133406-78-bin.zip压缩包到一个目录，得到解压后的目录结构如下图：

进入bin目录，发现有win32和win64两个文件夹，这2个文件夹分别对应windows32位和windows64位操作系统的启动脚本。

看自己的系统位数进入，会看到如下目录

其中activemq.bat便是启动脚本，双击启动。

ActiveMQ默认启动到8161端口，启动完了后在浏览器地址栏输入：http://localhost:8161/admin要求输入用户名密码，默认用户名密码为admin、admin，这个用户名密码是在conf/users.properties中配置的。输入用户名密码后便可看到如下图的ActiveMQ控制台界面了。

 3、ActiveMQ Linux 安装

• 安装环境

           1.需要jdk

           2.安装Linux系统。生产环境都是Linux系统。

• 安装步骤

     第一步： 把ActiveMQ 的压缩包上传到Linux系统。

              第二步：解压缩。

               第三步：启动。

       使用bin目录下的activemq命令启动：

                  [root@localhost bin]# ./activemq start

          关闭：

              [root@localhost bin]# ./activemq stop

          查看状态：

               [root@localhost bin]# ./activemq status

               //修改主机名

               Vim /etc/hosts

                

            //查看主机名

            

      注意：如果ActiveMQ整合spring使用不要使用activemq-all-5.12.0.jar包。建议使用5.11.2

           进入管理后台：http://192.168.25.168:8161/admin用户名：admin密码：admin

 4、ActiveMQ控制台介绍

Number Of Consumers   消费者 这个是消费者端的消费者数量 
Number Of Pending Messages   等待消费的消息 这个是当前未出队列的数量。可以理解为总接收数-总出队列数 
Messages Enqueued     进入队列的消息  进入队列的总数量,包括出队列的。 这个数量只增不减 
Messages Dequeued    出了队列的消息  可以理解为是消费这消费掉的数量 
     这个要分两种情况理解 
             在queues里它和进入队列的总数量相等(因为一个消息只会被成功消费一次),如果暂时不等是因为消费者还没来得及消费。 
            在 topics里 它因为多消费者从而导致数量会比入队列数高。 
    简单的理解上面的意思就是 

•  当有一个消息进入这个队列时，等待消费的消息是1，进入队列的消息是1。 
•  当消息消费后，等待消费的消息是0，进入队列的消息是1，出队列的消息是1. 
•  在来一条消息时，等待消费的消息是1，进入队列的消息就是2. 
•  没有消费者时  Pending Messages   和 入队列数量一样 
•  有消费者消费的时候 Pedding会减少 出队列会增加 
•   到最后 就是 入队列和出队列的数量一样多 
•   以此类推，进入队列的消息和出队列的消息是池子，等待消费的消息是水流。 

 5、ActiveMQ使用方法

         5.1 Queue

• Producer         

                  生产者：生产消息，发送端。 生产者生产了一个消息，只能由一个消费者进行消费

         pom文件引入依赖

 <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-core</artifactId>
            <version>5.7.0</version>
 </dependency>

      步骤:

     第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。

     第二步：使用ConnectionFactory对象创建一个Connection对象。

     第三步：开启连接，调用Connection对象的start方法。

     第四步：使用Connection对象创建一个Session对象。

     第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个Queue对象。

     第六步：使用Session对象创建一个Producer对象。

     第七步：创建一个Message对象，创建一个TextMessage对象。

     第八步：使用Producer对象发送消息。

              第九步：关闭资源。

public class provider {
    private  static String queue = "yehui1";
    public static void main(String[] args) throws JMSException {
        // 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
        //brokerURL服务器的ip及端口号
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        // 第二步：使用ConnectionFactory对象创建一个Connection对象。
        Connection connection = factory.createConnection();
        // 第三步：开启连接，调用Connection对象的start方法。
        connection.start();
        // 第四步：使用Connection对象创建一个Session对象。
        //第一个参数：是否开启事务。true：开启事务，第二个参数忽略。
        //第二个参数：当第一个参数为false时，才有意义。消息的应答模式。1、自动应答2、手动应答。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个Queue对象。
        //参数：队列的名称。
        Destination destination = session.createQueue(queue);
        // 第六步：使用Session对象创建一个Producer对象。
        MessageProducer producer = session.createProducer(destination);
        for (int i=0;i<10;i++) {
            // 第七步：创建一个Message对象，创建一个TextMessage对象。
            TextMessage textMessage = session.createTextMessage("消费者你好我来了" + i);
            // 第八步：使用Producer对象发送消息。
            producer.send(textMessage);
        }
        System.out.println("生产结束");
        // 第九步：关闭资源。
        producer.close();
        session.close();
        connection.close();
    }
}

• Consumer       

             消费者：接收消息。

               步骤:

         第一步：创建一个ConnectionFactory对象。

                   第二步：从ConnectionFactory对象中获得一个Connection对象。

                  第三步：开启连接。调用Connection对象的start方法。

                  第四步：使用Connection对象创建一个Session对象。

                  第五步：使用Session对象创建一个Destination对象。和发送端保持一致queue，并且队列的名称一致。

                  第六步：使用Session对象创建一个Consumer对象。

                  第七步：接收消息。

                  第八步：打印消息。

                  第九步：关闭资源

public class Comsoner {
    private  static String queue = "yehui1";

    public static void main(String[] args) throws JMSException, IOException {
        System.out.println("消费结束");
        // 第一步：创建一个ConnectionFactory对象。
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        // 第二步：从ConnectionFactory对象中获得一个Connection对象。
        Connection connection = factory.createConnection();
        // 第三步：开启连接。调用Connection对象的start方法。
        connection.start();
        // 第四步：使用Connection对象创建一个Session对象。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session对象创建一个Destination对象。和发送端保持一致queue，并且队列的名称一致。
        Queue queue = session.createQueue(Comsoner.queue);
        // 第六步：使用Session对象创建一个Consumer对象。
        MessageConsumer consumer = session.createConsumer(queue);
        // 第七步：接收消息。
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage) message;
                try {
                    //取消息的内容
                    String text = textMessage.getText();
                    System.out.println(text);
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        //等待键盘输入
        System.in.read();
        // 第九步：关闭资源
        consumer.close();
        session.close();
        connection.close();
    }
}

 5.2 Topic 

          Topic 话题，生产者生产了一个消息，可以由多个消费者进行消费

•  Producer

              使用步骤：

                     第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。

                     第二步：使用ConnectionFactory对象创建一个Connection对象。

                     第三步：开启连接，调用Connection对象的start方法。

                     第四步：使用Connection对象创建一个Session对象。

                     第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个Topic对象。

                      第六步：使用Session对象创建一个Producer对象。

                       第七步：创建一个Message对象，创建一个TextMessage对象。

                      第八步：使用Producer对象发送消息。

                     第九步：关闭资源。

public class Provider {

    private static String queue = "yehui-topic";
    public static void main(String[] args) throws JMSException {
        //第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
        // brokerURL服务器的ip及端口号
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        // 第二步：使用ConnectionFactory对象创建一个Connection对象。
        Connection connection = factory.createConnection();
        // 第三步：开启连接，调用Connection对象的start方法。
        connection.start();
        // 第四步：使用Connection对象创建一个Session对象。
        // 第一个参数：是否开启事务。true：开启事务，第二个参数忽略。
        // 第二个参数：当第一个参数为false时，才有意义。消息的应答模式。1、自动应答2、手动应答。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个topic对象。
        // 参数：话题的名称。
        Destination destination = session.createTopic(queue);
        // 第六步：使用Session对象创建一个Producer对象。
        MessageProducer producer = session.createProducer(destination);
        for (int i=0;i<10;i++) {
            // 第七步：创建一个Message对象，创建一个TextMessage对象。
            TextMessage textMessage = session.createTextMessage("消费者你好我来了" + i);
            // 第八步：使用Producer对象发送消息。
            producer.send(textMessage);
        }
        // 第九步：关闭资源。
        producer.close();
        session.close();
        connection.close();
    }
    }

• Consumer

          消费者：接收消息。

        使用步骤：

     第一步：创建一个ConnectionFactory对象。

     第二步：从ConnectionFactory对象中获得一个Connection对象。

     第三步：开启连接。调用Connection对象的start方法。

     第四步：使用Connection对象创建一个Session对象。

     第五步：使用Session对象创建一个Destination对象。和发送端保持一致topic，并且话题的名称一致。

     第六步：使用Session对象创建一个Consumer对象。

              第七步：接收消息。

              第八步：打印消息。

             第九步：关闭资源

public class Comsoner {
    private  static String queue = "yehui-topic";

    public static void main(String[] args) throws JMSException, IOException {
        //第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
        // brokerURL服务器的ip及端口号
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        // 第二步：使用ConnectionFactory对象创建一个Connection对象。
        Connection connection = factory.createConnection();
        // 第三步：开启连接，调用Connection对象的start方法。
        connection.start();
        // 第四步：使用Connection对象创建一个Session对象。
        // 第一个参数：是否开启事务。true：开启事务，第二个参数忽略。
        // 第二个参数：当第一个参数为false时，才有意义。消息的应答模式。1、自动应答2、手动应答。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session对象创建一个Destination对象。和发送端保持一致topic，并且话题的名称一致。
        Topic topic = session.createTopic(Comsoner.queue);
        // 第六步：使用Session对象创建一个Consumer对象。
        MessageConsumer consumer = session.createConsumer(topic);
        // 第七步：接收消息。
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage) message;
                try {
                    // 取消息的内容
                    String text = textMessage.getText();
                    System.out.println(text);
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        System.out.println("topic的消费端03。。。。。");
        // 等待键盘输入
        System.in.read();
        // 第九步：关闭资源
        consumer.close();
        session.close();
        connection.close();
    }
}

6、JMS消息可靠机制

ActiveMQ消息签收机制：

       客戶端成功接收一条消息的标志是一条消息被签收，成功应答。

消息的签收情形分两种：

       1、带事务的session

                如果session带有事务，并且事务成功提交，则消息被自动签收。如果事务回滚，则消息会被再次传送。

       2、不带事务的session

            不带事务的session的签收方式，取决于session的配置。

   Activemq支持一下三種模式：

   Session.AUTO_ACKNOWLEDGE  消息自动签收

   Session.CLIENT_ACKNOWLEDGE  客戶端调用acknowledge方法手动签收

    textMessage.acknowledge();//手动签收

   Session.DUPS_OK_ACKNOWLEDGE 不是必须签收，消息可能会重复发送。在第二次重新传送消息的时候，消息只有在被确认之后，

才认为已经被成功地消费了。消息的成功消费通常包含三个阶段：客户接收消息、客户处理消息和消息被确认。 在事务性会话中，当一个事务被提交的时候，确认自动发生。在非事务性会话中，消息何时被确认取决于创建会话时的应答模式（acknowledgement mode）。该参数有以下三个可选值：

Number Of Consumers  消费者 这个是消费者端的消费者数量 
Number Of Pending Messages 等待消费的消息 这个是当前未出队列的数量。可以理解为总接收数-总出队列数 
Messages Enqueued 进入队列的消息  进入队列的总数量,包括出队列的。 这个数量只增不减 
Messages Dequeued 出了队列的消息  可以理解为是消费这消费掉的数量 

代码案例:

provider 

public class provider {
    private  static String queue = "yehui1";
    public static void main(String[] args) throws JMSException {
        // 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
        //brokerURL服务器的ip及端口号
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        // 第二步：使用ConnectionFactory对象创建一个Connection对象。
        Connection connection = factory.createConnection();
        // 第三步：开启连接，调用Connection对象的start方法。
        connection.start();
        // 第四步：使用Connection对象创建一个Session对象。
        //第一个参数：是否开启事务。true：开启事务，第二个参数忽略。
        //第二个参数：当第一个参数为false时，才有意义。消息的应答模式。1、自动应答2、手动应答。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个Queue对象。
        //参数：队列的名称。
        Destination destination = session.createQueue(queue);
        // 第六步：使用Session对象创建一个Producer对象。
        MessageProducer producer = session.createProducer(destination);
        for (int i=0;i<10;i++) {
            // 第七步：创建一个Message对象，创建一个TextMessage对象。
            TextMessage textMessage = session.createTextMessage("消费者你好我来了" + i);
            // 第八步：使用Producer对象发送消息。
            producer.send(textMessage);
        }
        System.out.println("生产结束");
        // 第九步：关闭资源。
        producer.close();
        session.close();
        connection.close();
    }
}

Comsoner

public class Comsoner {
    private  static String queue = "yehui1";

    public static void main(String[] args) throws JMSException, IOException {
        System.out.println("消费结束");
        // 第一步：创建一个ConnectionFactory对象。
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        // 第二步：从ConnectionFactory对象中获得一个Connection对象。
        Connection connection = factory.createConnection();
        // 第三步：开启连接。调用Connection对象的start方法。
        connection.start();
        // 第四步：使用Connection对象创建一个Session对象。手动签收
        Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
        // 第五步：使用Session对象创建一个Destination对象。和发送端保持一致queue，并且队列的名称一致。
        Queue queue = session.createQueue(Comsoner.queue);
        // 第六步：使用Session对象创建一个Consumer对象。
        MessageConsumer consumer = session.createConsumer(queue);
        // 第七步：接收消息。
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage) message;
                try {
                    //取消息的内容
                    String text = textMessage.getText();
                    System.out.println(text);
                    //需要手动签收  如何这里不手动签收，将不会消费
                    textMessage.acknowledge();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        //等待键盘输入
        System.in.read();
        // 第九步：关闭资源
        consumer.close();
        session.close();
        connection.close();
    }
}

7、JMS可靠消息机制-持久话机制

PERSISTENT：指示JMS provider持久保存消息，以保证消息不会因为JMS provider的失败而丢失

NON_PERSISTENT:不要求JMS provider持久保存消息

// 设置消息持久化 producer.setDeliveryMode(DeliveryMode.PERSISTENT);

8、整合Eamil实现异步发送邮件信息

采用MQ方式发送方邮件原理图:

整合Eamil实现异步发送邮件信息

采用qq接口邮件接口

qq授权码申请：

  生产者:

   Maven依赖:

 <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-parent</artifactId>
        <version>2.0.6.RELEASE</version>
    </parent>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-activemq</artifactId>
        </dependency>
        <!-- fastjson 依赖 -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.47</version>
        </dependency>

    </dependencies>

yml文件

spring:
  activemq:
    broker-url: tcp://127.0.0.1:61616
    user: admin
    password: admin
queue: springboot-queue
server:
  port: 8080

config类

@Configuration
public class QueueConfig {

    @Value("${queue}")
    public String queue;

    @Bean
    public Queue getQueue(){
        return new ActiveMQQueue(queue);
    }

}

发送邮件代码

@RestController
public class QueueController {

    @Autowired
    private Destination queue;

    @Autowired
    private JmsMessagingTemplate messagingTemplate;

    @RequestMapping("/index")
    public String index(){
        messagingTemplate.convertAndSend(queue,"测试消息队列");
        return "index";
    }

    /**
     * 发送邮件代码
     * @return
     */
    @RequestMapping("/sendEmail")
    public String sendEmail(){
        JSONObject jsonObject = new JSONObject();
        String userName = System.currentTimeMillis() + "";
        jsonObject.put("userName", userName);
        jsonObject.put("eamil", "邮箱账号");
        String msg = jsonObject.toJSONString();
        System.out.println("生产者向消费者发送内容:" + msg);
        messagingTemplate.convertAndSend(queue, msg);
        return "sendEmail";
    }
}

启动类:

@SpringBootApplication
public class AppStart {
    public static void main(String[] args) {
        SpringApplication.run(AppStart.class);
    }
}

消费者:

  maven依赖

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-parent</artifactId>
        <version>2.0.6.RELEASE</version>
    </parent>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!--activemq依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-activemq</artifactId>
        </dependency>
        <!-- https://mvnrepository.com/artifact/com.alibaba/fastjson -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.47</version>
        </dependency>
        <!--邮件代码-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-mail</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
        </dependency>
    </dependencies>

yml文件

spring:
  activemq:
    broker-url: tcp://127.0.0.1:61616
    user: admin
    password: admin
  #邮件配置
  mail:
    #邮箱协议
    host: smtp.qq.com
    #用户账号
    username: 邮箱账号
    #授权码
    password: 邮箱授权码
    enable:  true
    smtp:
      auth: true
    starttls:
      enable: true
      required: true

queue: springboot-queue
server:
  port: 8081

发送邮件代码

@Component
public class Consumer {

    @Autowired
    private JavaMailSender javaMailSender;

    @JmsListener(destination = "${queue}")
    public void receive(String msg){
        if(StringUtils.isBlank(msg)){
            return;
        }
        sendSimpleMail(msg);
        System.out.println("监听器收到msg:" + JSONObject.parseObject(msg).toJSONString());
    }

    public void sendSimpleMail(String msg) {
        try {
            JSONObject jsonObject = JSONObject.parseObject(msg);
            String userName = jsonObject.getString("userName");
            String eamil = jsonObject.getString("eamil");
            SimpleMailMessage simpleMailMessage = new SimpleMailMessage();
            //邮件发送人
            simpleMailMessage.setFrom(eamil);
            //邮件接收人
            simpleMailMessage.setTo(eamil);
            //邮件主题
            simpleMailMessage.setSubject("邮件测试");
            //邮件内容
            simpleMailMessage.setText("邮件测试内容");
            javaMailSender.send(simpleMailMessage);
        } catch (Exception e) {
            System.out.println("邮件发送失败");
        }
    }
}

启动类

@SpringBootApplication
public class App {

    public static void main(String[] args) {
        SpringApplication.run(App.class);
    }
}