springboot集成ativemq

背景&＃xff1a;

       如果你想在springboot中使用activemq&＃xff0c;你可以有两个选择&＃xff0c;第一个就是使用内置的activemq&＃xff0c;还可以使用外部的activemq&＃xff0c;我在这里主要使用了外部的activemq&＃xff0c;我是在windows下部署的activemq&＃xff0c;不要忘记先启动它。

第一步&＃xff1a;添加maven依赖

org.springframework.bootspring-boot-starter-activemq


org.apache.activemqactivemq-pool


第二步&＃xff1a;配置applicaiton.properties配置文件

#配置activemq的连接地址
spring.activemq.broker-url&＃61;tcp://127.0.0.1:61616
#配置用户名
spring.activemq.user&＃61;admin
#配置密码
spring.activemq.password&＃61;admin
#true表示使用连接池&＃xff0c;false表示不使用连接池;当为false时&＃xff0c;每发送一条数据创建一个连接
spring.activemq.pool.enabled&＃61;true
#连接池最大连接数
spring.activemq.pool.max-connections&＃61;10
#空闲的连接过期时间&＃xff0c;默认为30s
spring.activemq.pool.idle-timeout&＃61;30000
#强制的连接过期时间&＃xff0c;与idleTimeout的区别在于&＃xff1a;idleTimeout是在连接空闲一段时间失效&＃xff0c;而expiryTimeout不管当前连接的情况&＃xff0c;只要达到指定时间就失效。默认为0&＃xff0c;never
spring.activemq.pool.expiry-timeout&＃61;0

第三步&＃xff1a;编写相应代码

       1、我把生产者和消费者都集成在了一个项目中&＃xff0c;项目结构图如下所示&＃xff1a;

       2、定义activemq的主题类型和名称&＃xff0c;我们这里定义了主题模式&＃xff08;名称为activemq.topic&＃xff09;和队列模式&＃xff08;名称为activemq.queue&＃xff09;

import javax.jms.Queue;
import javax.jms.Topic;import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;&＃64;Configuration
public class ActivemqConfig {// 定义队列模式&＃64;Beanpublic Queue queue1() {return new ActiveMQQueue("activemq.queue");}//定义主题模式&＃64;Beanpublic Topic topic() {return new ActiveMQTopic("activemq.topic");}}

       3、定义一个消息的提供者MessageProvider&＃xff0c;使用一个定时任务来持续的发送数据&＃xff0c;这里提供两种消息的发送方式&＃xff0c;使用哪一种方式都可以&＃xff0c;它们之间的差异性我们后续会分析&＃xff0c;方式一&＃xff08;强烈不推荐&＃xff09;&＃xff1a;

&＃64;Configuration
public class MessageProvider1 {&＃64;Autowiredprivate JmsMessagingTemplate jmsMessagingTemplate;&＃64;Scheduled(fixedRate&＃61;5000,initialDelay&＃61;3000)public void send() {// 向队列发送消息jmsMessagingTemplate.convertAndSend("activemq.queue", Math.random());// 向主题发送消息jmsMessagingTemplate.convertAndSend("activemq.topic", Math.random());}
}

           下面是方式二&＃xff0c;此时可以发现系统并没有指定queue的名称和topic的名称。

&＃64;Configuration
public class MessageProvider2 {&＃64;Autowiredprivate JmsMessagingTemplate jmsMessagingTemplate;&＃64;Autowiredprivate Queue queue;&＃64;Autowiredprivate Topic topic;&＃64;Scheduled(fixedRate&＃61;5000,initialDelay&＃61;3000)public void send() {// 向队列发送消息jmsMessagingTemplate.convertAndSend(queue, Math.random()); // 向主题发送消息jmsMessagingTemplate.convertAndSend(topic, Math.random());}
}

       4、定义一个消息的消费者MessageCustomer

&＃64;Component
public class MessageCustomer {&＃64;JmsListener(destination &＃61; "activemq.queue")public void q1(String text) {System.out.println("消费者consumer0消费的消息为&＃xff1a;"&＃43;text);}&＃64;JmsListener(destination &＃61; "activemq.queue")public void q2(String text) {System.out.println("消费者consumer1消费的消息为&＃xff1a;"&＃43;text);}&＃64;JmsListener(destination &＃61; "activemq.topic")public void t1(String text) {System.out.println("消费者consumer2消费的消息为&＃xff1a;"&＃43;text);}&＃64;JmsListener(destination &＃61; "activemq.topic")public void t2(String text) {System.out.println("消费者consumer3消费的消息为&＃xff1a;"&＃43;text);}
}

       5、定义启动类Application

&＃64;SpringBootApplication
&＃64;EnableScheduling
// 启动消息队列
&＃64;EnableJms
public class Application {public static void main(String[] args) {SpringApplication.run(Application.class, args);}
}

第四步&＃xff1a;使用MessageProvider2进行测试

       1、我们首先采用MessageProvider2的方式进行测试&＃xff0c;启动程序&＃xff0c;输出结果如下所示&＃xff0c;我们发现输出结果并不是我们想要的那种结果&＃xff0c;如果是我们想要的结果输出结果还得打印消费者consumer2和消费者consumer3消费的消息。

      2、结果分析&＃xff1a;原来springboot默认采用的是p2p模式&＃xff08;队列&＃xff09;进行消息的监听。所以无法监听到topic模式的消息&＃xff0c;那么如何对topic模式的消息进行监听呢&＃xff1f;只需要在applicaiton.properties里面加上这样一个属性&＃xff1a;

spring.jms.pub-sub-domain&＃61;true

       3、重启项目&＃xff0c;输出结果如下图所示&＃xff0c;此时topic模式的消息正常消费&＃xff0c;但是队列模式的消息又无法正常消费&＃xff0c;原因是springboot默认只支持一种类型消息的消费&＃xff0c;那么如何支持两种呢&＃xff1f;我们后面会说。

第五步&＃xff1a;使用MessageProvider1进行测试

       1、我们首先注释掉刚才加在application.properties里面加的spring.jms.pub-sub-domain&＃61;true&＃xff0c;启动项目&＃xff0c;输出结果如下所示&＃xff1a;

       2、我们发现输出结果并不是我们想要的&＃xff0c;springboot默认支持队列模式的消息监听&＃xff0c;但是当我门发送消息时&＃xff0c;只是指定了消息接收方的名字&＃xff0c;并没有指定消息接收方的类型&＃xff0c;所以springboot默认将这两个消息提供者都看成了队列类型。所以才会出现现在的结果。

       3、在application.properties里面加上spring.jms.pub-sub-domain&＃61;true&＃xff0c;重启项目&＃xff0c;输出结果如下所示&＃xff1a;

       4、我们发现只是将队列类型改成了主题类型&＃xff0c;从根本上还是没有解决我们的问题&＃xff0c;我们想让springboot既支持队列模式又支持主题模式。 

第六步&＃xff1a;springboot同时支持topic和queue

       1、注释掉在application.properties里面加的spring.jms.pub-sub-domain&＃61;true

       2、在ActivemqConfig这个类里面重新定义两个bean

&＃64;Configuration
public class ActivemqConfig {// 定义队列模式&＃64;Beanpublic Queue queue() {return new ActiveMQQueue("activemq.queue");}//定义主题模式&＃64;Beanpublic Topic topic() {return new ActiveMQTopic("activemq.topic");}&＃64;Beanpublic JmsListenerContainerFactory topicListenerContainerFactory(ConnectionFactory connectionFactory){DefaultJmsListenerContainerFactory factory &＃61; new DefaultJmsListenerContainerFactory();factory.setPubSubDomain(true);factory.setConnectionFactory(connectionFactory);return factory;}&＃64;Beanpublic JmsListenerContainerFactory queueListenerContainerFactory(ConnectionFactory connectionFactory){DefaultJmsListenerContainerFactory factory &＃61; new DefaultJmsListenerContainerFactory();factory.setPubSubDomain(false);factory.setConnectionFactory(connectionFactory);return factory;}
}

       3、在MessageCustomer的方法注解JmsListener上添加属性containerFactory属性

&＃64;Component
public class MessageCustomer {&＃64;JmsListener(destination &＃61; "activemq.queue",containerFactory &＃61; "queueListenerContainerFactory")public void q1(String text) {System.out.println("消费者consumer0消费的消息为&＃xff1a;"&＃43;text); }&＃64;JmsListener(destination &＃61; "activemq.queue",containerFactory &＃61; "queueListenerContainerFactory") public void q2(String text) {System.out.println("消费者consumer1消费的消息为&＃xff1a;"&＃43;text); }&＃64;JmsListener(destination &＃61; "activemq.topic",containerFactory &＃61; "topicListenerContainerFactory")public void t1(String text) {System.out.println("消费者consumer2消费的消息为&＃xff1a;"&＃43;text);}&＃64;JmsListener(destination &＃61; "activemq.topic",containerFactory &＃61; "topicListenerContainerFactory")public void t2(String text) {System.out.println("消费者consumer3消费的消息为&＃xff1a;"&＃43;text);}
}

       4、重启项目&＃xff0c;输出结果如下&＃xff0c;我们发现现在的结果就是同时支持了queue模式和topic模式。

第七步&＃xff1a;遇坑分析

        1、一直以为在activemq的管理界面中点击Queues里面显示的创建了多少个queue&＃xff08;队列&＃xff09;模式的主题&＃xff0c;这个理解是错误的。其实这里面显示的信息是“MessageCustomer里面监听&＃xff08;destination&＃xff09;的&＃xff08;所有监听的主题和类型都会出现在这里&＃xff0c;不论它是队列模式还是主题模式&＃xff09;&＃43;你自己创建的队列类型的主题”。

       我们在destination中监听了activemq.queue和activemq.topic这两个&＃xff0c;但我们创建了activemq.queue2队列和activemq.topic2主题&＃xff0c;并向它们分别发送了消息&＃xff0c;注意&＃xff1a;如果不向activemq.queue2和activemq.topic2发送消息&＃xff0c;仅仅是定义了一个bean&＃xff0c;那么activemq就发现不了它们两个。此时的Queues的组成就是监听的activemq.queue和activemq.topic&＃43;创建的activemq.queue2。

       2、如果我们想看我们定义的topic&＃xff0c;进入到Topics下查看即可。