springboot与ActiveMQ整合

ゞ浴缸里的玫瑰 2022-03-18 08:58 181阅读 0赞

## 前言 ##

很多项目, 都不是一个系统就做完了. 而是好多个系统, 相互协作来完成功能. 那, 系统与系统之间, 不可能完全独立吧?

如: 在学校所用的管理系统中, 有学生系统, 资产系统, 宿舍系统等等. 当学期结束之后, 是否需要对已经结束的期次进行归档操作. 假如归档功能在学生系统中, 那点击归档之后, 学生是不是还要关心宿舍那边是否已结束, 学生所领资产是否全都归还?

显然, 这并不是一个好的方式, 系统之间的耦合性做的太强了, 很不利于系统扩展, 而且, 一步操作, 可能要等很久很久, 才能完成. 用户可愿意等?

既然同步归档不可能了, 那是否有办法实现异步归档? 异步归档怎么实现呢?

我们其实可以通过消息队列来实现异步归档. 学生这边点击归档后, 发个消息到队列中, 其他系统自行去读取, 然后完成各自系统应该完成的工作.

## ActiveMQ下载安装 ##

下载地址: http://activemq.apache.org/download.html

安装过程比较简单, 在centos中, 解压出来, 就算是安装好了

运行方法:

![640690-20180224123108276-650806528.png][]

运行起来后, 可以通过 ip:8161 来查看是否成功.

![640690-20180224125247546-2097228680.png][]

点击红框中的链接, 会出现登录弹框, 账号密码默认都是admin.

![640690-20180224125403078-1969812703.png][]

## springboot整合activemq ##

** 一. 目录结构**

![640690-20180224123327661-1955389331.png][]

producer : 消息生产者

consumer-a : 消息消费者

consumer-b : 消息消费者

pom文件:

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-activemq</artifactId>
    </dependency>

如果使用pool的话, 就需要在pom中加入以下依赖:

<dependency>
         <groupId>org.apache.activemq</groupId>
         <artifactId>activemq-pool</artifactId>
         <version>5.14.5</version>
    </dependency>

** 二. producer**

1. 目录结构

![640690-20180224123759634-41515991.png][]

2. yml文件:

server:
      port: 8080
      context-path: /pro
    spring:
      activemq:
        user: admin
        password: admin
        broker-url: tcp://192.168.153.129:61616
        pool:
          enabled: true
          max-connections: 10
    
    queueName: publish.queue
    topicName: publish.topic

这里我开启了连接池, 默认是不开的.

这里要注意端口, 不是之前的8161.

2. 配置文件  ActiveMQConfig

@Configuration
    public class ActiveMQConfig {
        @Value("${queueName}")
        private String queueName;
    
        @Value("${topicName}")
        private String topicName;
    
        @Value("${spring.activemq.user}")
        private String usrName;
    
        @Value("${spring.activemq.password}")
        private  String password;
    
        @Value("${spring.activemq.broker-url}")
        private  String brokerUrl;
    
        @Bean
        public Queue queue(){
            return new ActiveMQQueue(queueName);
        }
    
        @Bean
        public Topic topic(){
            return new ActiveMQTopic(topicName);
        }
    
        @Bean
        public ActiveMQConnectionFactory connectionFactory() {
            return new ActiveMQConnectionFactory(usrName, password, brokerUrl);
        }
    
        @Bean
        public JmsListenerContainerFactory<?> jmsListenerContainerQueue(ActiveMQConnectionFactory connectionFactory){
            DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
            bean.setConnectionFactory(connectionFactory);
            return bean;
        }
    
        @Bean
        public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ActiveMQConnectionFactory connectionFactory){
            DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
            //设置为发布订阅方式, 默认情况下使用的生产消费者方式
            bean.setPubSubDomain(true);
            bean.setConnectionFactory(connectionFactory);
            return bean;
        }
    }

这里本来不需要配置这么多的, 但是在consumer中也会用到, 所以就暂时弄一份一样的, 拷贝一下完事.

3. PublishController

@RestController
    @RequestMapping("/publish")
    public class PublishController {
    
        @Autowired
        private JmsMessagingTemplate jms;
    
        @Autowired
        private Queue queue;
    
        @Autowired
        private Topic topic;
    
        @RequestMapping("/queue")
        public String queue(){
    
            for (int i = 0; i < 10 ; i++){
                jms.convertAndSend(queue, "queue"+i);
            }
    
            return "queue 发送成功";
        }
    
        @JmsListener(destination = "out.queue")
        public void consumerMsg(String msg){
            System.out.println(msg);
        }
    
        @RequestMapping("/topic")
        public String topic(){
    
            for (int i = 0; i < 10 ; i++){
                jms.convertAndSend(topic, "topic"+i);
            }
    
            return "topic 发送成功";
        }
    }

** 三. consumer **

1. 目录结构

![640690-20180224124510167-142513057.png][]

a,b是一样的, 只是显示的信息不同.

2. 配置文件

yml配置文件是一样的, 只是修改了端口和context-path.

ActiveMQConfig文件内容是一样的.

3. listener

@Component
    public class QueueListener {
    
        @JmsListener(destination = "publish.queue", containerFactory = "jmsListenerContainerQueue")
        @SendTo("out.queue")
        public String receive(String text){
            System.out.println("QueueListener: consumer-a 收到一条信息: " + text);
            return "consumer-a received : " + text;
        }
    }

SendTo 会将此方法返回的数据, 写入到 queue : out.queue 中去.

@Component
    public class TopicListener {
    
        @JmsListener(destination = "publish.topic", containerFactory = "jmsListenerContainerTopic")
        public void receive(String text){
            System.out.println("TopicListener: consumer-a 收到一条信息: " + text);
        }
    }

这里通过传入不同的factory, 来实现发送不同类型的信息.

四. 测试

1. queue测试

浏览器中访问: http://localhost:8080/pro/publish/queue

![640690-20180224125809579-1903850360.png][]

然后看一下, 控制台, 那些用户接收到了信息.

![640690-20180224125935799-308871732.png][]

![640690-20180224125946663-693631297.png][]

从上两幅图看的出来, a, b并不能同时接收数据. 这是queue的方式, 点对点.

![640690-20180224130044871-1970557652.png][]

那我想点对面, 怎么办?

2. topic测试

浏览器访问页面: http://localhost:8080/pro/publish/topic

![640690-20180224130206283-783010989.png][]

![640690-20180224130228958-2090908948.png][]

a用户完全接收到信息了. 再看看b用户

![640690-20180224130251053-1183544804.png][]

没毛病, 也都接收到数据了.

topic默认情况下, 是不会保存数据的, 也就是说, consumer是接收不到之前未接收到的信息.

而queue却是可以的.

但是, topic并不是不能实现那个功能, 只要配置一下, 还是可以的.

推荐链接

[消息中间件企业级应用][Link 1]

[640690-20180224123108276-650806528.png]: /images/20220318/44c889035fe94f83af4b2015aa62e438.png
[640690-20180224125247546-2097228680.png]: /images/20220318/0a845d5d5e654ae8968c39a227587db1.png
[640690-20180224125403078-1969812703.png]: /images/20220318/b8ddeac5a35d498c9be8b5c5b5c3509d.png
[640690-20180224123327661-1955389331.png]: /images/20220318/fa2fa8474ac743f28f1efa5c0dfa318e.png
[640690-20180224123759634-41515991.png]: /images/20220318/d8b5cfb982214fcab70e04a51d774ae7.png
[640690-20180224124510167-142513057.png]: /images/20220318/bb4353cb75224ae7901379722c851bde.png
[640690-20180224125809579-1903850360.png]: /images/20220318/97a8d72c296847279e2d211114263f83.png
[640690-20180224125935799-308871732.png]: /images/20220318/73ecfd16ecc44b03a53f16c4c2f693e1.png
[640690-20180224125946663-693631297.png]: /images/20220318/74300e7b59be496f92e3a71d50d08688.png
[640690-20180224130044871-1970557652.png]: /images/20220318/cdb9b8a31546487393acf3c66b2b8598.png
[640690-20180224130206283-783010989.png]: /images/20220318/fa68f17a80d742f28ec5c549505c43a6.png
[640690-20180224130228958-2090908948.png]: /images/20220318/8a22fe3023ff4a8ea00caa1e6ce5e05f.png
[640690-20180224130251053-1183544804.png]: /images/20220318/cd4526f54a314905a5f58b375561c1e0.png
[Link 1]: https://www.javazhiyin.com/31897.html