Springboot整合ActiveMQ（Queue和Topic两种模式）

水深无声 2023-07-19 10:35 9阅读 0赞

> 写在前面： 从2018年底开始学习SpringBoot，也用SpringBoot写过一些项目。这里对学习Springboot的一些知识总结记录一下。如果你也在学习SpringBoot，可以关注我，一起学习，一起进步。

**相关文章：**

[【Springboot系列】Springboot入门到项目实战][Springboot_Springboot]

--------------------

### 文章目录 ###

*   *  ActiveMQ简介
     *   *  1、ActiveMQ简介
         *  2、ActiveMQ下载
     *  SpringBoot整合ActiveMQ
     *   *  1、新建SpringBoot项目
         *  2、项目结构
         *  3、相关配置信息
         *  4、ActiveMQ配置类
     *  Queue队列模式
     *   *  1、队列生产者
         *  2、队列消费者
         *  3、测试效果
     *  Topic模式
     *   *  1、topic生产者
         *  2、topic消费者
         *  3、测试效果

--------------------

## ActiveMQ简介 ##

### 1、ActiveMQ简介 ###

Apache ActiveMQ是Apache软件基金会所研发的开放源代码消息中间件；由于ActiveMQ是一个纯Java程序，因此只需要操作系统支持Java虚拟机，ActiveMQ便可执行。

### 2、ActiveMQ下载 ###

下载地址：[http://activemq.apache.org/components/classic/download/][http_activemq.apache.org_components_classic_download]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70]  
下载完成后解压双击activemq.bat文件打开(不用安装，直接使用)，目录和打开后效果如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70 2]  
运行后，浏览器访问http://localhost:8161/地址进入一下界面。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70 3]  
点击Manage ActiveMQ broker登录到ActiveMQ管理页面，默认账号和密码都是admin。管理页面如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70 4]

## SpringBoot整合ActiveMQ ##

### 1、新建SpringBoot项目 ###

新建Springboot项目，添加对应的依赖。项目完整的pom.xml文件如下：

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.2.5.RELEASE</version>
            <relativePath/> 
        </parent>
        <groupId>com.mcy</groupId>
        <artifactId>springboot-mq</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>springboot-mq</name>
        <description>Demo project for Spring Boot</description>
    
        <properties>
            <java.version>1.8</java.version>
        </properties>
    
        <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>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
                <exclusions>
                    <exclusion>
                        <groupId>org.junit.vintage</groupId>
                        <artifactId>junit-vintage-engine</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    </project>

### 2、项目结构 ###

### 3、相关配置信息 ###

在application.properties类中添加ActiveMQ相关的配置信息

server.port=8080
    server.servlet.context-path=/mq
    
    #MQ服务器地址
    spring.activemq.broker-url=tcp://localhost:61616
    #用户名
    spring.activemq.user=admin
    #密码
    spring.activemq.password=admin
    #设置是Queue队列还是Topic，false为Queue，true为Topic，默认false-Queue
    spring.jms.pub-sub-domain=false
    #spring.jms.pub-sub-domain=true
    
    #变量，定义队列和topic的名称
    myqueue: activemq-queue
    mytopic: activemq-topic

### 4、ActiveMQ配置类 ###

ActiveMQ配置类ConfigBean，配置了Queue队列和topic两种模式，代码如下：

import org.apache.activemq.command.ActiveMQQueue;
    import org.apache.activemq.command.ActiveMQTopic;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.jms.annotation.EnableJms;
    import org.springframework.stereotype.Component;
    import javax.jms.Topic;
    
    /**
     1. MQ配置类
     */
    @Component
    @EnableJms
    public class ConfigBean {
        @Value("${myqueue}")
        private String myQueue;
        @Value("${mytopic}")
        private String topicName;
    
        //队列
        @Bean
        public ActiveMQQueue queue(){
            return new ActiveMQQueue(myQueue);
        }
    
        //topic
        @Bean
        public Topic topic(){
            return new ActiveMQTopic(topicName);
        }
    }

## Queue队列模式 ##

队列模式即点对点传输。  
点对点消息传递域的特点如下：

1.  每个消息只能有一个消费者，类似于1对1的关系。好比个人快递自己领自己的。
2.  消息的生产者和消费者之间没有时间上的相关性。无论消费者在生产者发送消息的时候是否处于运行状态，消费者都可以提取消息。好比我们的发送短信，发送者发送后不见得接收者会即收即看。
3.  消息被消费后队列中不会再存储，所以消费者不会消费到已经被消费掉的消息。

### 1、队列生产者 ###

QueueProducerController类为队列生产者控制器，主要向消息队列中发送消息。代码如下：

import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsMessagingTemplate;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    import javax.jms.Queue;
    
    /*
     * 队列消息生产者
     */
    @RestController
    public class QueueProducerController {
        @Autowired
        private JmsMessagingTemplate jmsMessagingTemplate;
    
        @Autowired
        private Queue queue;
    
        /*
         * 消息生产者
         */
        @RequestMapping("/sendmsg")
        public void sendmsg(String msg) {
            System.out.println("发送消息到队列：" + msg);
            // 指定消息发送的目的地及内容
            this.jmsMessagingTemplate.convertAndSend(this.queue, msg);
        }
    }

### 2、队列消费者 ###

QueueConsumerController类为队列消费者控制器，具体代码如下：

import org.springframework.beans.factory.annotation.Value;
    import org.springframework.jms.annotation.JmsListener;
    import org.springframework.web.bind.annotation.RestController;
    
    /*
     1. 队列queue消费者控制器
     */
    @RestController
    public class QueueConsumerController {
        /*
         * 消费者接收消息
         */
        @JmsListener(destination="${myqueue}")
        public void readActiveQueue(String message) {
            System.out.println("接受到：" + message);
        }
    }

### 3、测试效果 ###

运行项目在浏览器中访问http://localhost:8080/mq/sendmsg?msg=123。向消息队列中发送123。控制台输出效果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70 5]  
**ActiveMQ控制台显示：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70 6]

1.  Number Of Pending Messages：消息队列中待处理的消息
2.  Number Of Consumers：消费者的数量
3.  Messages Enqueued：累计进入过消息队列的总量
4.  Messages Dequeued：累计消费过的消息总量

**【注】队列模式时，配置文件application.properties中spring.jms.pub-sub-domain属性必须设置为false。**

## Topic模式 ##

topic模式基于发布/订阅模式的传输。  
基于发布/订阅模式的传输的特点如下：

1.  生产者将消息发布到topic中，每个消息可以有多个消费者，属于1：N的关系；
2.  生产者和消费者之间有时间上的相关性。订阅某一个主题的消费者只能消费自它订阅之后发布的消息。
3.  生产者生产时，topic不保存消息它是无状态的不落地，假如无人订阅就去生产，那就是一条废消息。

### 1、topic生产者 ###

TopicProducerController类为topic生产者控制器，主要向消息队列中发送消息。代码如下：

import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsMessagingTemplate;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    import javax.jms.Queue;
    import javax.jms.Topic;
    
    /*
    * topic消息生产者
    */
    @RestController
    public class TopicProducerController {
       @Autowired
       private JmsMessagingTemplate jmsMessagingTemplate;
    
       @Autowired
       private Topic topic;
    
       /*
        * 消息生产者
        */
       @RequestMapping("/topicsendmsg")
       public void sendmsg(String msg) {
           System.out.println("发送消息到MQ：" + msg);
           // 指定消息发送的目的地及内容
           this.jmsMessagingTemplate.convertAndSend(this.topic, msg);
       }
    }

### 2、topic消费者 ###

TopicConsumerController类为topic消费者控制器，其中写了两个消费者方法，可以理解为有两个用户订阅。具体代码如下：

import org.springframework.jms.annotation.JmsListener;
    import org.springframework.web.bind.annotation.RestController;
    
    /*
     1. topic消费者控制器
     */
    @RestController
    public class TopicConsumerController {
        /*
         * 消费者接收消息
         */
        @JmsListener(destination="${mytopic}")
        public void readActiveQueue(String message) {
            System.out.println("接受到：" + message);
        }
    
        @JmsListener(destination="${mytopic}")
        public void readActiveQueue1(String message) {
            System.out.println("接受到：" + message);
        }
    }

### 3、测试效果 ###

运行项目在浏览器中访问http://localhost:8080/mq/topicsendmsg?msg=123。向消息队列中发送123。控制台输出效果（有两个消费者方法）：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70 7]  
**ActiveMQ控制台显示：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70 8]

1.  Number Of Consumers：消费者的数量
2.  Messages Enqueued：累计进入过消息队列的总量
3.  Messages Dequeued：累计消费过的消息总量

**【注】Topic模式时，配置文件application.properties中spring.jms.pub-sub-domain属性必须设置为true。**

> 最后有什么不足之处，欢迎大家指出，期待与你的交流。如果感觉对你有帮助，点个赞在走呗 O(∩\_∩)O ~

[Springboot_Springboot]: https://blog.csdn.net/qq_40205116/article/details/104269651
[http_activemq.apache.org_components_classic_download]: http://activemq.apache.org/components/classic/download/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70]: /images/20230528/667a24fd374e45d89f76275a87a73bd6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70 1]: /images/20230528/356e901127e24754adb77d4f4c8052f2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70 2]: /images/20230528/3d5cb7556f814f53b6a1c86527f36852.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70 3]: /images/20230528/f2f1f15cca0a4042acc6712664d524c6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70 4]: /images/20230528/ba60e42fb5ac4e65b58f627b2b2f26c7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70 5]: /images/20230528/dc0ee6ce8f9a41828f6aa2d71b9c8940.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70 6]: /images/20230528/dc5cb0e7aae74926a4db94027772cc22.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70 7]: /images/20230528/423131179a2a41dc8135a66d0f1bef01.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjA1MTE2_size_16_color_FFFFFF_t_70 8]: /images/20230528/f367b317460e49d88545e2b1249bbd17.png