ActiveMq虚拟主题+SpringBoot

布满荆棘的人生 2021-09-02 09:58 356阅读 0赞

【前情回顾】：最近在做项目的过程中，需要用到虚拟主题。**本文以Activemq为主**。一般情况下主题的使用都是订阅一个mq的name，然后对这个name进行监听，做业务逻辑处理。但是之前从没听过虚拟主题，后来一番了解之后，特地总结一下。

【简介】:

> ActiveMQ中，topic只有在持久订阅下才是持久化的。持久订阅时，每个持久订阅者，都相当于一个queue的客户端，它会收取所有消息。这种情况下存在两个问题：   
> 1：同一应用内consumer端负载均衡的问题：也即是同一个应用上的一个持久订阅不能使用多个consumer来共同承担消息处理功能。因为每个consumer都会获取所有消息。   
> queue模式可以解决这个问题，但broker端又不能将消息发送到多个应用端。所以，既要发布订阅，又要让消费者分组，这个功能JMS规范本身是没有的。   
> 2：同一应用内consumer端failover的问题：由于只能使用单个的持久订阅者，如果这个订阅者出错，则应用就无法处理消息了，系统的健壮性不高 
> 
> 为了解决这两个问题，ActiveMQ中实现了虚拟Topic的功能。

【虚拟主题的使用】：

1.创建

> 对于消息发布者来说，正常的Topic，名称以VirtualTopic.开头。比如现有一个虚拟的topic的name为：VirtualTopic.myTopic。使用的时候命名为：Consumer.XXX.VirtualTopic.myTopic。

public class VirtualTopic {
        public void createVirtualTopic () {
            ConnectionFactory connectionFactory = new
                    ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
            Connection connection = connectionFactory.createConnection();
            Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createTopic("VirtualTopic.myTopic");
            MessageProducer producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);//设置传送模式为持久模式，默认为非持久
            connection.start();
            for(int i=0; i<3; i++) {
                TextMessage message = session.createTextMessage("message--"+i);
                Thread.sleep(1000);
                //通过消息生产者发出消息
                producer.send(message);
            }
            session.commit();
            session.close();
            connection.close();
        }
    }

2.监听

> 对于消息接收端来说，是个队列，不同应用里使用不同的前缀作为队列的名称，即可表明自己的身份即可实现消费端应用分组。本文重在对虚拟主题的使用，结合SpringBoot框架。具体步骤如下：
> 
> 1.  编写配置文件。
> 2.  创建queue。
> 3.  监听对应的queue。

@Configuration
    public class ListenerContainerConfig {
        //对应application.properties中activemq的url
        @Value("${spring.activemq.broker-url}")
        private String url;
        //对应application.properties中activemq的user
        @Value("${spring.activemq.user}")
        private String username;
        //对应application.properties中activemq的password
        @Value("${spring.activemq.password}")
        private String password;
    
        @Bean("queueListenerContainer")
        public JmsListenerContainerFactory queueListenerContainer() {
            DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
            //监听的是Queue
            bean.setPubSubDomain(false);
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(username, password, url);
            bean.setConnectionFactory(connectionFactory);
            return bean;
        }
    }

创建一个name为：Consumer.A.VirtualTopic.myTopic

//在初始化的时候创建queue
    @Bean
        public void createVirtualTopic(){
            logger.info("虚拟主题初始化...");
            new ActiveMQQueue("Consumer.A.VirtualTopic.myTopic");
        }

监听对应的queue:

@JmsListener(destination = "Consumer.A.VirtualTopic.myTopic", containerFactory = "queueListenerContainer")
        public void subscribeHeatMatchEvent(String content) {
            //接收报文，对事件进行JSON字符串解析。
            //进行业务逻辑操作
            
        }

【扩展】：自定义更改消费虚拟地址

> 默认虚拟主题的前缀是 ：VirtualTopic.\*。自定义消费虚拟地址默认格式：Consumer.\*.VirtualTopic.\*。  
> 自定义消费虚拟地址可以改，比如下面的配置就把它修改了。  
> xml配置示例如下：  
> <broker xmlns="http://activemq.apache.org/schema/core">  
>     <destinationInterceptors>  
>         <virtualDestinationInterceptor>  
>             <virtualDestinations>  
>                 <virtualTopic name=">" prefix="VirtualTopicConsumers.\*." selectorAware="false"/>  
>             </virtualDestinations>  
>         </virtualDestinationInterceptor>  
>     </destinationInterceptors>  
> </broker>