SpringBoot使用ActiveMq同时支持点对点推送和发布订阅

憧憬博客

发布于 2020-07-21 07:08:38

1.3K00

代码可运行

文章被收录于专栏：憧憬博客分享憧憬博客分享

运行总次数：0

代码可运行

在SpringBoot中使用ActiveMq默认是只能点对点推送, ActiveMq还有一种方式就是发布订阅, 一个发布者, 多个订阅者, 形成一个点对面

先来配置一下点对面的。

application.properties 增加配置

#default point to point 开启发布订阅
spring.jms.pub-sub-domain=true

xxApplication.java

// 将对象交给spring来管理
@Bean
    public Topic topic(){
        return new ActiveMQTopic("common.topic");
    }

inteface ProducerService.java 增加

/**
 * 发布消息
 * @param message
 */
public void publish(final String message);

ProducerServiceImpl.java 增加

@Autowired
private Topic topic;

@Override
public void publish(String message) {
    jmsMessagingTemplate.convertAndSend(topic, message);
}

订阅者 CommonTopicSub.java

package com.example.demo.jms;

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

@Component
public class CommonTopicSub {

    @JmsListener(destination = "common.topic")
    public void receiveTopic1(String text) {
        System.out.println("common.topic1 接受到的消息 ====" + text);
    }

    @JmsListener(destination = "common.topic")
    public void receiveTopic2(String text) {
        System.out.println("common.topic2 接受到的消息 ====" + text);
    }
}

controller

@GetMapping("publish")
public Object publish(String msg) {
    producerService.publish(msg);
    return JsonUtils.buildSuccess("ok");
}

效果

这样就完成了我们的发布订阅, 但是测试的时候发现点对点推送不好用, 消息开始堆积, 我们需要让它同时支持两种

默认消费者并不会消费订阅发布类型的消息，这是由于springboot默认采用的是p2p模式进行消息的监听

在配置文件里面，注释掉 #spring.jms.pub-sub-domain=true

修改 CommonTopicSub.java

/**
 * @ JmsListener如果不指定独立的containerFactory的话是只能消费queue消息 是订阅不到消息的
 * @param text
 */
@JmsListener(destination = "common.topic", containerFactory="jmsListenerContainerTopic")
public void receiveTopic1(String text) {
    System.out.println("common.topic1 接受到的消息 ====" + text);
}

xxApplication.java 增加

@Bean
public JmsListenerContainerFactory<?> jmsListenerContainerTopic() {
    DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
    bean.setPubSubDomain(true);
    bean.setConnectionFactory(connectionFactory()); // 创建连接的方法 如果不知道看上篇快速入门文章
    return bean;
}