SpringBoot+ActiveMQ-发布订阅模式（生产端）

小咪咪 2024-03-27 09:26 62阅读 0赞

**SpringBoot+ActiveMQ-发布订阅模式（生产端）**

**Topic 主题**

**\* 消息消费者（订阅方式）消费该消息**

**\* 消费生产者将发布到topic中，同时有多个消息消费者（订阅）消费该消息**

**\* 这种方式和点对点方式不同，发布到topic的消息会被所有订阅者消费**

**\* 当生产者发布消息，不管是否有消费者，都不会保存消息，如果对订阅消息提前做了持久化操作，还是可以收到的**

ActiveMQ版本：apache-activemq-5.16.5

案例源码:[SpringBoot+ActiveMQ-发布订阅Demo][SpringBoot_ActiveMQ-_Demo]

SpringBoot集成ActiveMQ topic生产端的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.topic.provider</groupId>
        <artifactId>boot-example-topic-demo-provider-2.0.5</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>boot-example-topic-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=8043
     
     
    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
    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

启动类AppTopicProvider

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

ActiveMqConfig

package boot.example.topic.provider.config;
    
    import org.apache.activemq.command.ActiveMQTopic;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.jms.annotation.EnableJms;
    import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
    import org.springframework.jms.config.JmsListenerContainerFactory;
    
    import javax.jms.ConnectionFactory;
    import javax.jms.Topic;
    
    /**
     * 蚂蚁舞
     */
    
    @Configuration
    public class ActiveMqConfig {
    
        public static final String defaultTopic = "myw_topic";
    
        @Bean
        public Topic topic() {
            return new ActiveMQTopic(defaultTopic);
        }
    
    
    //    // topic模式的ListenerContainer
    //    @Bean
    //    public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ConnectionFactory activeMQConnectionFactory) {
    //        DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
    //        bean.setPubSubDomain(true);
    //        bean.setConnectionFactory(activeMQConnectionFactory);
    //        return bean;
    //    }
    
    }

ProviderDefaultTopicService

package boot.example.topic.provider.service;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsMessagingTemplate;
    import org.springframework.stereotype.Service;
    
    import javax.jms.Topic;
    
    
    @Service
    public class ProviderDefaultTopicService {
    
        @Autowired
        private Topic topic;
    
        @Autowired
        private JmsMessagingTemplate jmsMessagingTemplate;
    
        /**
         *  使用默认bean配置的名称发送数据
         */
        public void sendStringDefaultTopic(String message) {
            this.jmsMessagingTemplate.convertAndSend(topic, message);
        }
    
    
        
    }

ProviderTopicService

package boot.example.topic.provider.service;
    
    import boot.example.queue.entity.BootProvider;
    import org.apache.activemq.command.ActiveMQTopic;
    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 ProviderTopicService {
    
    
        @Autowired
        private JmsMessagingTemplate jmsMessagingTemplate;
    
        /**
         * 发送字符串消息主题
         *
         * @param topicName 主题名称
         * @param message   字符串
         */
        public void sendStringTopic(String topicName, String message) {
            this.jmsMessagingTemplate.convertAndSend(new ActiveMQTopic(topicName), message);
        }
    
        /**
         * 发送字符串集合消息主题
         *
         * @param topicName 主题名称
         * @param list      字符串集合
         */
        public void sendStringListTopic(String topicName, List<String> list) {
            this.jmsMessagingTemplate.convertAndSend(new ActiveMQTopic(topicName), list);
        }
    
        /**
         * 发送对象消息主题
         *
         * @param topicName 主题名称
         * @param obj       对象
         */
        public void sendObjTopic(String topicName, BootProvider obj) {
            this.jmsMessagingTemplate.convertAndSend(new ActiveMQTopic(topicName), obj);
        }
    
        /**
         * 发送对象集合消息主题
         *
         * @param topicName 主题名称
         * @param objList   对象集合
         */
        public void sendObjListTopic(String topicName, List<BootProvider> objList) {
            this.jmsMessagingTemplate.convertAndSend(new ActiveMQTopic(topicName), objList);
        }
    
    }

BootDefaultTopicProviderController

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

BootTopicProviderController

package boot.example.topic.provider.controller;
    
    import boot.example.queue.entity.BootProvider;
    import boot.example.topic.provider.service.ProviderTopicService;
    import org.springframework.web.bind.annotation.*;
    
    import javax.annotation.Resource;
    import java.util.List;
    
    /**
     *     蚂蚁舞
     */
    @RestController
    @RequestMapping(value="/provider")
    public class BootTopicProviderController {
    
        //  指定Topic
        public static final String stringTopic = "stringTopic";
    
        //  指定list<String>
        public static final String stringListTopic = "stringListTopic";
    
        //  指定Object
        public static final String objTopic = "objTopic";
    
        //  指定List<Object>
        public static final String objListTopic = "objListTopic";
    
    
        @Resource
        private ProviderTopicService providerTopicService;
    
        @PostMapping(value = "/sendStringTopic")
        @ResponseBody
        public String sendStringTopic(@RequestParam(name="message",required = true) String message) throws Exception {
            providerTopicService.sendStringTopic(stringTopic, message);
            return "success";
        }
    
        @PostMapping(value = "/sendStringListTopic")
        @ResponseBody
        public String sendStringListTopic(@RequestBody List<String> list) throws Exception {
            if(list.isEmpty()){return "fail";}
            providerTopicService.sendStringListTopic(stringListTopic, list);
            return "success";
        }
    
        @PostMapping(value = "/sendObjTopic")
        @ResponseBody
        public String sendObjTopic(@RequestBody BootProvider bootProvider) throws Exception {
            if(bootProvider == null){return "fail";}
            providerTopicService.sendObjTopic(objTopic, bootProvider);
            return "success";
        }
    
        @PostMapping(value = "/sendObjListTopic")
        @ResponseBody
        public String sendObjListTopic(@RequestBody  List<BootProvider> list) throws Exception {
            if(list.isEmpty()){return "fail";}
            providerTopicService.sendObjListTopic(objListTopic, list);
            return "success";
        }
    
    
    
    
    
    }

SwaggerConfig UI界面测试用

package boot.example.topic.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  地址和端口根据实际项目查看
         */
    
    
    }

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-topic-demo-provider-2.0.5
    │  │  pom.xml
    │  ├─src
    │  │  ├─main
    │  │  │  ├─java
    │  │  │  │  └─boot
    │  │  │  │      └─example
    │  │  │  │          └─topic
    │  │  │  │              └─provider
    │  │  │  │                  │  AppTopicProvider.java
    │  │  │  │                  │  
    │  │  │  │                  ├─config
    │  │  │  │                  │      ActiveMqConfig.java
    │  │  │  │                  │      SwaggerConfig.java
    │  │  │  │                  │      
    │  │  │  │                  ├─controller
    │  │  │  │                  │      BootDefaultTopicProviderController.java
    │  │  │  │                  │      BootTopicProviderController.java
    │  │  │  │                  │      
    │  │  │  │                  └─service
    │  │  │  │                          ProviderDefaultTopicService.java
    │  │  │  │                          ProviderTopicService.java
    │  │  │  │                          
    │  │  │  └─resources
    │  │  │          application.properties
    │  │  │          
    │  │  └─test
    │  │      └─java
    │  │          └─boot
    │  │              └─example
    │  │                  └─topic
    │  │                      └─provider
    │  │                              AppTopicProviderTest.java

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

http://localhost:8043/doc.html

![8c39b8466ace4b44b9af774b5929916d.png][]

![d93d3cfea3624169a6f490d9719e02a8.png][]

SpringBoot集成activeMQ对于队列(queue)来说是持久化的事物模式，因此拿来即用，但是发布订阅（topic主题）模式不是持久化的，发送端把消息发出后，如果没有消费者，消息不会被消费，即使消费者端启动了，也不会消费之前的消息(除非对发布订阅也做持久化)

**典型的案例topic demo测试（支持持久化订阅）**

ActiveMQTopicProvider(Topic生产端)

package boot.example.topic.provider;
    
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /**
     *  蚂蚁舞
     */
    public class ActiveMQTopicProvider {
    
        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开启 false不开启
            //第二个参数：消息是否自动确认  acknowledgeMode
            //当transacted为true时，acknowledgeMode会默认Session.SESSION_TRANSACTED
            Session session = connection.createSession(true, Session.DUPS_OK_ACKNOWLEDGE);
            //创建Topic
            Topic topic = session.createTopic("myw_topic_test");
            MessageProducer producer = session.createProducer(topic);
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);//持久化设置 默认就是这个
            Message message = session.createTextMessage("myyhtw蚂蚁也会跳舞");
            producer.send(message);
    
            session.commit();  // 开启事务必须提交这个
            producer.close();
            session.close();
            connection.close();
        }
    }

ActiveMQTopicConsumer(Topic消费端)

package boot.example.topic.provider;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /**
     * 蚂蚁舞
     */
    public class ActiveMQTopicConsumer {
        public static void main(String[] args) throws Exception {
    
            String clientId = "myw_topic_test_wijwe";
            //创建连接工厂
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "tcp://127.0.0.1:61616");
            //  持久订阅的Client端标识(多个端持久订阅需要保证唯一性，否则可能会出现问题)
            connectionFactory.setClientID(clientId);
    
            //创建连接
            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);
            //创建topic
            Topic topic = session.createTopic("myw_topic_test");
            //持久订阅
    
            //创建消费者
    //        // 普通订阅 topic不是持久化的
    //        MessageConsumer consumer = session.createConsumer(topic);
            // 持久订阅
            MessageConsumer consumer = session.createDurableSubscriber(topic,clientId);
    
            while(true){
                //失效时间，如果10秒内没有收到新的消息，说明没有消息存在，此时可以退出当前循环
                TextMessage message = (TextMessage) consumer.receive(60000);
                if(message!=null){
                    System.out.println(message.getText());
                    //message.acknowledge();
                } else {
                    break;
                }
            }
    
            //关闭连接
            session.commit();
            session.close();
            connection.close();
        }
    }

ActiveMQTopicConsumerListener（Topic消费端监听）

package boot.example.topic.provider;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /**
     * 蚂蚁舞
     */
    public class ActiveMQTopicConsumerListener {
    
    
        public static void main(String[] args) throws Exception {
            String clientId = "myw_topic_test_www2";
            //创建连接工厂
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "tcp://127.0.0.1:61616");
            //创建连接
            Connection connection = connectionFactory.createConnection();
            //  持久订阅的Client端标识(多个端持久订阅需要保证唯一性，否则可能会出现问题)
            connection.setClientID(clientId);
            //开启连接
            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);
            //创建topic
            Topic topic = session.createTopic("myw_topic_test");
            //创建消费者
    //        // 普通订阅 topic不是持久化的
    //        MessageConsumer consumer = session.createConsumer(topic);
            // 持久订阅
            MessageConsumer consumer = session.createDurableSubscriber(topic,clientId);
    
            //注册消息监听器
            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();
        }
    
    }

topic持久化需要发布端设置持久化

producer.setDeliveryMode(DeliveryMode.PERSISTENT);//持久化设置 默认就是这个

topic消费订阅端在代码里有这几个设置

String clientId = "myw_topic_test_www2";
    
    //  持久订阅的Client端标识(多个端持久订阅需要保证唯一性，否则可能会出现问题)
    connection.setClientID(clientId);
    
    // 持久订阅
    MessageConsumer consumer = session.createDurableSubscriber(topic,clientId);

[SpringBoot_ActiveMQ-_Demo]: https://download.csdn.net/download/myyhtw/87399206
[8c39b8466ace4b44b9af774b5929916d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/e2af19ade22644ea8895177cd9aa2610.png
[d93d3cfea3624169a6f490d9719e02a8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/207563553d6748be9d1d8ead4a4a3841.png