SpringBoot+ActiveMQ-点对点队列模式（生产端）

以你之姓@ 2024-03-27 09:22 78阅读 0赞

ActiveMQ消息中间件的点对点模式*point to point 消息队列*

**\* 消息消费者从queue中取出并且消费消息**

**\* 消息被消费以后，queue中不再有存储，所以消息消费者不可能消费到已经被消费的消息**

**\* queue中支持存在多个消费者，但是对一个消费者而言，只会有一个消费者可以消费，其他的消费者便不能消费此消息**

**\* 当消费者不存在时，消息会一直保存，直到有消费者消费消息 (持久化)**

ActiveMQ环境配置地址：[ActiveMQ在Centos下的安装][ActiveMQ_Centos]

ActiveMQ版本：apache-activemq-5.16.5

SpringBoot集成ActiveMQ生产端的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 http://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.0.5.RELEASE</version>
            <relativePath/> 
        </parent>
    
        <groupId>boot.example.queue.provider</groupId>
        <artifactId>boot-example-queue-demo-provider-2.0.5</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>boot-example-queue-demo-provider-2.0.5</name>
        <description>Demo project for Spring Boot</description>
    
        <properties>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
            <java.version>1.8</java.version>
        </properties>
    
        <dependencies>
    
            <dependency>
                <groupId>boot.example.demo.entity</groupId>
                <artifactId>boot-example-demo-entity-2.0.5</artifactId>
                <version>0.0.1-SNAPSHOT</version>
            </dependency>
    
            <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>
                <exclusions>
                    <exclusion>
                        <groupId>org.slf4j</groupId>
                        <artifactId>slf4j-log4j12</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
            
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-pool</artifactId>
                <version>5.16.5</version>
            </dependency>
            <dependency>
                <groupId>io.springfox</groupId>
                <artifactId>springfox-swagger2</artifactId>
                <version>2.9.2</version>
            </dependency>
            <dependency>
                <groupId>com.github.xiaoymin</groupId>
                <artifactId>swagger-bootstrap-ui</artifactId>
                <version>1.9.2</version>
            </dependency>
    
    
    
    
    
    
    
        </dependencies>
    
        <build>
            <plugins>
                
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                    <executions>
                        <execution>
                            <goals>
                                <goal>repackage</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>
            </plugins>
        </build>
    
    </project>

application.properties

server.port=8041
    
    
    spring.activemq.broker-url=tcp://127.0.0.1:61616
    spring.activemq.user=admin
    spring.activemq.password=admin
    spring.activemq.in-memory=false
    spring.activemq.packages.trust-all=true
    #true连接池
    spring.activemq.pool.enabled=true
    spring.activemq.pool.max-connections=6
    spring.activemq.pool.idle-timeout=30000
    spring.activemq.pool.expire-timeout=0
    spring.jms.pub-sub-domain=false

启动类AppQueueProvider

package boot.example.queue.provider;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.jms.annotation.EnableJms;
    
    /**
     * 蚂蚁舞
     */
    @SpringBootApplication
    /**
     * 启动消息队列
     */
    @EnableJms
    public class AppQueueProvider {
        public static void main( String[] args )
        {
            SpringApplication.run(AppQueueProvider.class, args);
            System.out.println( "Hello World!" );
        }
        
    }

配置一个queue ActiveMQQueueConfig

package boot.example.queue.provider.config;
    
    
    import org.apache.activemq.command.ActiveMQQueue;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import javax.jms.Queue;
    
    
    /**
     *  蚂蚁舞
     */
    @Configuration
    public class ActiveMQQueueConfig {
    
        public static final String defaultQueue = "myw_queue";
    
        @Bean
        public Queue queue() {
            return new ActiveMQQueue(defaultQueue);
        }
    
    }

BootDefaultProviderService

package boot.example.queue.provider.service;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsMessagingTemplate;
    import org.springframework.stereotype.Service;
    
    import javax.jms.Queue;
    
    /**
     *  蚂蚁舞
     */
    @Service
    public class BootDefaultProviderService {
    
        @Autowired
        private Queue queue;
    
        @Autowired
        private JmsMessagingTemplate jmsMessagingTemplate;
    
        public void sendStringDefaultQueue(String message) {
            jmsMessagingTemplate.convertAndSend(queue, message);
        }
    
    
    }

其他模式BootProviderService

package boot.example.queue.provider.service;
    
    import boot.example.queue.entity.BootProvider;
    import org.apache.activemq.command.ActiveMQQueue;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsMessagingTemplate;
    import org.springframework.stereotype.Service;
    
    import java.util.List;
    
    /**
     *  蚂蚁舞
     */
    @Service
    public class BootProviderService {
    
        @Autowired
        private JmsMessagingTemplate jmsMessagingTemplate;
    
        /**
         * 发送字符串消息队列
         *
         * @param queueName 队列名称
         * @param message   字符串
         */
        public void sendStringQueue(String queueName, String message) {
            jmsMessagingTemplate.convertAndSend(new ActiveMQQueue(queueName), message);
        }
    
        /**
         * 发送字符串集合消息队列
         *
         * @param queueName 队列名称
         * @param list      字符串集合
         */
        public void sendStringListQueue(String queueName, List<String> list) {
            jmsMessagingTemplate.convertAndSend(new ActiveMQQueue(queueName), list);
        }
    
        /**
         * 发送对象消息队列
         *
         * @param queueName 队列名称
         * @param obj       对象
         */
        public void sendObjQueue(String queueName, BootProvider obj) {
            jmsMessagingTemplate.convertAndSend(new ActiveMQQueue(queueName), obj);
        }
    
        /**
         * 发送对象集合消息队列
         *
         * @param queueName 队列名称
         * @param objList   对象集合
         */
        public void sendObjListQueue(String queueName, List<BootProvider> objList) {
            jmsMessagingTemplate.convertAndSend(new ActiveMQQueue(queueName), objList);
        }
    
    }

使用SwaggerConfig

package boot.example.queue.provider.config;
    
    import com.google.common.base.Predicates;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import springfox.documentation.builders.ApiInfoBuilder;
    import springfox.documentation.builders.PathSelectors;
    import springfox.documentation.builders.RequestHandlerSelectors;
    import springfox.documentation.service.ApiInfo;
    import springfox.documentation.spi.DocumentationType;
    import springfox.documentation.spring.web.plugins.Docket;
    import springfox.documentation.swagger2.annotations.EnableSwagger2;
    
    /**
     *  蚂蚁舞
     */
    @Configuration
    @EnableSwagger2
    public class SwaggerConfig {
    
        @Bean
        public Docket createRestApi(){
            return new Docket(DocumentationType.SWAGGER_2).apiInfo(apiInfo()).select()
                    .apis(RequestHandlerSelectors.any()).paths(PathSelectors.any())
                    .paths(Predicates.not(PathSelectors.regex("/error.*")))
                    .paths(PathSelectors.regex("/.*"))
                    .build().apiInfo(apiInfo());
        }
    
        private ApiInfo apiInfo(){
            return new ApiInfoBuilder()
                    .title("demo")
                    .description("demo接口")
                    .version("0.01")
                    .build();
        }
    
        /**
         * http://localhost:XXXX/doc.html  地址和端口根据实际项目查看
         */
    
    
    }

BootQueueDefaultProviderController

package boot.example.queue.provider.controller;
    
    import boot.example.queue.provider.service.BootDefaultProviderService;
    import org.springframework.web.bind.annotation.*;
    
    import javax.annotation.Resource;
    
    /**
     *     蚂蚁舞
     */
    @RestController
    @RequestMapping(value="/provider")
    public class BootQueueDefaultProviderController {
    
        @Resource
        private BootDefaultProviderService providerService;
    
        @PostMapping(value = "/sendStringDefaultQueue")
        @ResponseBody
        public String sendStringDefaultQueue(@RequestParam(name="message",required = true) String message) throws Exception {
            providerService.sendStringDefaultQueue(message);
            return "success";
        }
    
    
    
    
    
    }

BootQueueProviderController

package boot.example.queue.provider.controller;
    
    import boot.example.queue.entity.BootProvider;
    import boot.example.queue.provider.service.BootProviderService;
    import org.springframework.web.bind.annotation.*;
    import javax.annotation.Resource;
    import java.util.List;
    
    /**
     *     蚂蚁舞
     */
    @RestController
    @RequestMapping(value="/provider")
    public class BootQueueProviderController {
    
        //  指定queue
        public static final String stringQueue = "stringQueue";
    
        //  指定list<String>
        public static final String stringListQueue = "stringListQueue";
    
        //  指定Object
        public static final String objQueue = "objQueue";
    
        //  指定List<Object>
        public static final String objListQueue = "objListQueue";
    
        @Resource
        private BootProviderService providerService;
    
        @PostMapping(value = "/sendStringQueue")
        @ResponseBody
        public String sendStringQueue(@RequestParam(name="message",required = true) String message) throws Exception {
            providerService.sendStringQueue(stringQueue, message);
            return "success";
        }
    
        @PostMapping(value = "/sendStringListQueue")
        @ResponseBody
        public String sendStringListQueue(@RequestBody List<String> list) throws Exception {
            if(list.isEmpty()){return "fail";}
            providerService.sendStringListQueue(stringListQueue, list);
            return "success";
        }
    
        @PostMapping(value = "/sendObjQueue")
        @ResponseBody
        public String sendObjQueue(@RequestBody BootProvider bootProvider) throws Exception {
            if(bootProvider == null){return "fail";}
            providerService.sendObjQueue(objQueue, bootProvider);
            return "success";
        }
    
        @PostMapping(value = "/sendObjListQueue")
        @ResponseBody
        public String sendObjListQueue(@RequestBody  List<BootProvider> list) throws Exception {
            if(list.isEmpty()){return "fail";}
            providerService.sendObjListQueue(objListQueue, list);
            return "success";
        }
    
    
    
    
    
    }

BootProvider对象

package boot.example.queue.entity;
    
    import java.io.Serializable;
    import java.util.Date;
    
    /**
     *  用在activeMq消息，必须保证package一致，不然序列化后反序列化要出错
     *  蚂蚁舞
     */
    public class BootProvider implements Serializable {
    
        private int id;
    
        private String name;
    
        private Date date = new Date();
    
        public BootProvider() {
        }
    
        public BootProvider(int id, String name) {
            this.id = id;
            this.name = name;
        }
    
        public int getId() {
            return id;
        }
    
        public void setId(int id) {
            this.id = id;
        }
    
        public String getName() {
            return name;
        }
    
        public void setName(String name) {
            this.name = name;
        }
    
        public Date getDate() {
            return date;
        }
    
        public void setDate(Date date) {
            this.date = date;
        }
    
        @Override
        public String toString() {
            return "BootProvider{" +
                    "id=" + id +
                    ", name='" + name + '\'' +
                    ", date=" + date +
                    '}';
        }
    }

代码结构

├─boot-example-demo-entity-2.0.5
    │  │  pom.xml
    │  │  
    │  ├─src
    │  │  └─main
    │  │      └─java
    │  │          └─boot
    │  │              └─example
    │  │                  └─queue
    │  │                      └─entity
    │  │                              BootProvider.java
    │  │                              
    ├─boot-example-queue-demo-provider-2.0.5
    │  │  pom.xml
    │  │  
    │  ├─src
    │  │  ├─main
    │  │  │  ├─java
    │  │  │  │  └─boot
    │  │  │  │      └─example
    │  │  │  │          └─queue
    │  │  │  │              └─provider
    │  │  │  │                  │  AppQueueProvider.java
    │  │  │  │                  │  
    │  │  │  │                  ├─config
    │  │  │  │                  │      ActiveMQQueueConfig.java
    │  │  │  │                  │      SwaggerConfig.java
    │  │  │  │                  │      
    │  │  │  │                  ├─controller
    │  │  │  │                  │      BootQueueDefaultProviderController.java
    │  │  │  │                  │      BootQueueProviderController.java
    │  │  │  │                  │      
    │  │  │  │                  └─service
    │  │  │  │                          BootDefaultProviderService.java
    │  │  │  │                          BootProviderService.java
    │  │  │  │                          
    │  │  │  └─resources
    │  │  │          application.properties
    │  │  │          
    │  │  └─test
    │  │      └─java
    │  │          └─boot
    │  │              └─example
    │  │                  └─queue
    │  │                      └─provider
    │  │                              ActiveMQQueueConsumer.java
    │  │                              ActiveMQQueueProvider.java
    │  │                              AppQueueProviderTest.java

ActiveMQ在SpringBoot里的代码demo集成完成，(ActiveMQ已启动后)启动程序访问

http://localhost:8041/doc.html

![99d68e95f34049ea95fa36fde4b99b88.png][]

测试发送可以看到发送成功了

![8557ed0f8b8b4a28bdb95b32c254efce.png][]

SpringBoot集成activeMQ默认是持久化的事物模式，因此拿来即用，完全不需要花时间去学习ActiveMQ更多的知识,刚需就是能够一个服务把消息发送到另一个服务，并且ActiveMQ支持事物和持久化

典型的案例demo测试

ActiveMQQueueProvider(生产端)

package boot.example.queue.provider;
    
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /**
     *  蚂蚁舞
     */
    public class ActiveMQQueueProvider {
    
        public static void main(String[] args) throws Exception{
            //1 创建连接工厂
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "tcp://127.0.0.1:61616");
            //2 创建连接
            Connection connection = connectionFactory.createConnection();
            //3 打开连接
            connection.start();
            //4 创建会话
            //第一个参数：是否开启事务 transacted true开启 false不开启
            //第二个参数：消息是否自动确认  acknowledgeMode
            //当transacted为true时，acknowledgeMode会默认Session.SESSION_TRANSACTED
            Session session = connection.createSession(true, Session.DUPS_OK_ACKNOWLEDGE);
            //创建队列
            Queue queue = session.createQueue("myw_queue_test");
            //5 创建生产者
            MessageProducer producer = session.createProducer(queue);
            //6 创建消息
            Message message = session.createTextMessage("myyhtw蚂蚁也会跳舞");
            //7 发送消息
            producer.send(message);
    
            //8 关闭消息
            session.commit();  // 开启事务必须提交这个
            producer.close();
            session.close();
            connection.close();
        }
    }

ActiveMQQueueConsumer(消费端)

package boot.example.queue.provider;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    public class ActiveMQQueueConsumer {
        public static void main(String[] args) throws Exception {
            //创建连接工厂
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "tcp://127.0.0.1:61616");
            //创建连接
            Connection connection = connectionFactory.createConnection();
            //开启连接
            connection.start();
            // 创建会话
            // transacted 如果设置true，操作消息队列后，必须使用 session.commit() 如果设置false，操作消息队列后，不使用session.commit();
            // acknowledgeMode
            // 1-Session.AUTO_ACKNOWLEDGE 自动应答
            // 2-Session.CLIENT_ACKNOWLEDGE 手动应答
            // 3-Session.DUPS_OK_ACKNOWLEDGE 延迟应答
            // 0-Session.SESSION_TRANSACTED 事务
            // 当transacted为true时，acknowledgeMode会默认Session.SESSION_TRANSACTED
            Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            //创建队列
            Queue queue = session.createQueue("myw_queue_test");
            //创建消费者
            MessageConsumer consumer = session.createConsumer(queue);
            while(true){
                //失效时间，如果10秒内没有收到新的消息，说明没有消息存在，此时可以退出当前循环
                TextMessage message = (TextMessage) consumer.receive(10000);
                if(message!=null){
                    System.out.println(message.getText());
                    //message.acknowledge();
                } else {
                    break;
                }
            }
    
            //关闭连接
            session.commit();
            session.close();
            connection.close();
        }
    }

ActiveMQQueueConsumerListener(消费端监听器)

package boot.example.queue.provider;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    public class ActiveMQQueueConsumerListener {
    
    
        public static void main(String[] args) throws Exception {
            //创建连接工厂
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "tcp://127.0.0.1:61616");
            //创建连接
            Connection connection = connectionFactory.createConnection();
            //开启连接
            connection.start();
            // 创建会话
            // transacted 如果设置true，操作消息队列后，必须使用 session.commit() 如果设置false，操作消息队列后，不使用session.commit();
            // acknowledgeMode
            // 1-Session.AUTO_ACKNOWLEDGE 自动应答
            // 2-Session.CLIENT_ACKNOWLEDGE 手动应答
            // 3-Session.DUPS_OK_ACKNOWLEDGE 延迟应答
            // 0-Session.SESSION_TRANSACTED 事务
            // 当transacted为true时，acknowledgeMode会默认Session.SESSION_TRANSACTED
            Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
            //创建队列
            Queue queue = session.createQueue("myw_queue_test");
            //创建消费者
            MessageConsumer consumer = session.createConsumer(queue);
    
            //注册消息监听器
            consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    try {
                        TextMessage msg = (TextMessage)message;
                        System.out.println(msg.getText());
                        message.acknowledge();
                    } catch (JMSException e) {
                        throw new RuntimeException(e);
                    }
                }
            });
    
            while (true){}
    //        consumer.close();
    //        session.close();
    //        connection.close();
        }
    
    }

**源码案例地址:** [ SpringBoot+ActiveMQ源码案例][SpringBoot_ActiveMQ]

[ActiveMQ_Centos]: https://blog.csdn.net/myyhtw/article/details/128762044
[99d68e95f34049ea95fa36fde4b99b88.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/e8f9c11f4bd644b5bd50c82af04c9d51.png
[8557ed0f8b8b4a28bdb95b32c254efce.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/4c759cf1b5854609889b0e3e250367aa.png
[SpringBoot_ActiveMQ]: https://download.csdn.net/download/myyhtw/87398483