Spring Boot 2.X - Spring Boot整合JMS之ActiveMQ（发布订阅模式）

￡神魔★判官ぃ 2022-02-19 17:55 338阅读 0赞

### 文章目录 ###

*  Spring Boot 2.X - Spring Boot整合JMS之ActiveMQ
 *   *  1.修改配置
     *  2.订阅者
     *  3.修改接口
     *  4.启动测试
     *  5.问题？
     *  6.点对点和发布订阅共存解决
     *   *  6.1 新建配置类
         *  6.2 修改配置
         *  6.3 修改订阅者
         *  6.4 启动测试
     *  7.项目代码

# Spring Boot 2.X - Spring Boot整合JMS之ActiveMQ #

> `Spring Boot 2`整合ActiveMQ案例之订阅发布方式。文中`Spring Boot`版本为`2.1.4.RELEASE`。

上篇 [Spring Boot 2.X - Spring Boot整合JMS之ActiveMQ][Spring Boot 2.X - Spring Boot_JMS_ActiveMQ]介绍了ActiveMQ点对点的模式，这里我们来看下订阅发布模式。

## 1.修改配置 ##

开启pub-sub模式，默认是关闭的，也就是默认是点对点模式：

spring:
      activemq:
        broker-url: tcp://192.168.0.2:61616 # activemq消息组件的连接主机
        user: admin
        password: admin # 账号密码默认为admin
      jms:
        pub-sub-domain: true # 启用发布订阅模式（默认关闭是点对点模式）

## 2.订阅者 ##

新建订阅者`TopicSub`

@Component
    public class TopicSub { 
    
        @JmsListener(destination = "topic.test"/*,containerFactory = "topicListenerContainerFactory"*/)
        public void receive1(String message) { 
            System.err.println("Topic.Subscribe.receive1接收消息：" + message);
        }
    
        @JmsListener(destination = "topic.test"/*,containerFactory = "topicListenerContainerFactory"*/)
        public void receive2(String message) { 
            System.err.println("Topic.Subscribe.receive2接收消息：" + message);
        }
    
        @JmsListener(destination = "topic.test"/*,containerFactory = "topicListenerContainerFactory"*/)
        public void receive3(String message) { 
            System.err.println("Topic.Subscribe.receive3接收消息：" + message);
        }
    
        @JmsListener(destination = "topic.test"/*,containerFactory = "topicListenerContainerFactory"*/)
        public void receive4(String message) { 
            System.err.println("Topic.Subscribe.receive4接收消息：" + message);
        }
    }

并在ActiveMQ管理后台增加一条名为`topic.test`的Topic，如下：  
![在这里插入图片描述][20190416213231226.png]

## 3.修改接口 ##

在`MsgController`中增加消息发布接口

@GetMapping("/sendTopic")
        public String sendTopic(String msg) { 
            jmsTemplate.convertAndSend(new ActiveMQTopic("topic.test"), msg);
            return "SUCCESS";
        }

## 4.启动测试 ##

启动ActiveMQ服务，并启动XxxApplication本项目，测试  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0UwOTYyMDEyNg_size_16_color_FFFFFF_t_70]

## 5.问题？ ##

这里有个问题，我们一下之前的点对点消息模式：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0UwOTYyMDEyNg_size_16_color_FFFFFF_t_70 1]

> 我们发现，并没有看到消费的信息打印出来。查阅ActiveMQ管理后台，有消息未被消费  
> ![在这里插入图片描述][20190416214119281.png]

## 6.点对点和发布订阅共存解决 ##

### 6.1 新建配置类 ###

新建配置类，并添加Bean：`JmsListenerContainerFactory`，设置监听容器支持Pub-Sub模式

@Configuration
    public class MyConfig { 
    
        @Bean
        public JmsListenerContainerFactory<?> topicListenerContainerFactory(ConnectionFactory connectionFactory) { 
            DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
            factory.setPubSubDomain(true);
            factory.setConnectionFactory(connectionFactory);
            return factory;
        }
    }

### 6.2 修改配置 ###

注释掉`spring.jms.pub-sub-domain`，或者设为`false`

spring:
      activemq:
        broker-url: tcp://192.168.0.2:61616 # activemq消息组件的连接主机
        user: admin
        password: admin # 账号密码默认为admin
    # jms:
    # pub-sub-domain: true # 启用发布订阅模式（默认关闭是点对点模式）

### 6.3 修改订阅者 ###

指定`containerFactory`为上一步设置的`topicListenerContainerFactory`。

@Component
    public class TopicSub { 
    
        @JmsListener(destination = "topic.test",containerFactory = "topicListenerContainerFactory")
        public void receive1(String message) { 
            System.err.println("Topic.Subscribe.receive1接收消息：" + message);
        }
    
        @JmsListener(destination = "topic.test",containerFactory = "topicListenerContainerFactory")
        public void receive2(String message) { 
            System.err.println("Topic.Subscribe.receive2接收消息：" + message);
        }
    
        @JmsListener(destination = "topic.test"/*,containerFactory = "topicListenerContainerFactory"*/)
        public void receive3(String message) { 
            System.err.println("Topic.Subscribe.receive3接收消息：" + message);
        }
    
        @JmsListener(destination = "topic.test"/*,containerFactory = "topicListenerContainerFactory"*/)
        public void receive4(String message) { 
            System.err.println("Topic.Subscribe.receive4接收消息：" + message);
        }
    }

### 6.4 启动测试 ###

启动ActiveMQ和本项目，

1.  测试点对点模式：![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0UwOTYyMDEyNg_size_16_color_FFFFFF_t_70 2]
2.  测试发布订阅模式：![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0UwOTYyMDEyNg_size_16_color_FFFFFF_t_70 3]  
    已经成功实现点对点和发布订阅模式共存了。

## 7.项目代码 ##

完整代码：[Github][]

[Spring Boot 2.X - Spring Boot_JMS_ActiveMQ]: https://blog.csdn.net/xudc0521/article/details/89333506
[20190416213231226.png]: /images/20220219/10fc90d170484303933c36f8bacfeb78.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0UwOTYyMDEyNg_size_16_color_FFFFFF_t_70]: /images/20220219/b5f136d438764cb891a6ad353a7319dc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0UwOTYyMDEyNg_size_16_color_FFFFFF_t_70 1]: /images/20220219/4bcd3b68f4da4d5babbe420696851836.png
[20190416214119281.png]: /images/20220219/8b0c8c9400584b2789dcd9e2247aa43e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0UwOTYyMDEyNg_size_16_color_FFFFFF_t_70 2]: /images/20220219/3ba2226510c14e91bdca02a490f2552b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0UwOTYyMDEyNg_size_16_color_FFFFFF_t_70 3]: /images/20220219/a3f18be05e1d4a6385f578be81cd1db1.png
[Github]: https://github.com/xudc0521/spring-boot-v2/tree/master/spring-boot-v2-jms-activemq