解決Springboot整合ActiveMQ發送和接收topic消息的問題

環境搭建

1.創建maven項目(jar)

2.pom.xml添加依賴

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.4.0.RELEASE</version>
    </parent>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-activemq</artifactId>
        </dependency>
    </dependencies>

 

3.編寫引導類

@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class,args);
    }
}

 

4.在resources下的application.properties配置文件中添加對應的配置

spring.activemq.broker-url=tcp://192.168.25.131:61616
#此url為activeMQ所在服務器的鏈接

 

5.在類中注入JmsMessageTemplate

@Autowired
private JmsMessagingTemplate jmsMessagingTemplate;

6.調用JmsMessageTemplate的方法發送消息

jmsMessagingTemplate.convertAndSend(“queue消息名稱”,"消息內容");

設置消息發送類型

在引導類中配置消息類型

Queue

@Bean(name="queue")
public Destination getQueue(){
    return new ActiveMQQueue("queue_test");
}

Topic

@Bean(name="topic")
public Destination getTopic(){
    return new ActiveMQTopic("topic_test");
}

類中注入發送消息類型

在發送消息的類中注入發送消息類型對象Destination

Queue

@Resource(name="queue")
private Destination queue;

Topic

@Autowired
@Qualifier(value="topic")
private Destination topic;

消息發送

jmsMessageTemplate.convertAndSend(queue,"消息內容");
jmsMessageTemplate.convertAndSend(topic,"消息內容");

編寫消費者

@Component
public class ActiveMQConsumer {
    //接收queue消息
    @JmsListener(destination = "queue_test")
    public void handler(String message){
        System.out.println(message);
    }
    //接收topic消息
    @JmsListener(destination = "topic_test")
    public void handlerTopic(String msessage){
        System.out.println(msessage);
    }
}

啟動測試

controller類的方法中添加
@RequestMapping("/send")
public void sendQueue(){
    jmsMessagingTemplate.convertAndSend(queue,"這是Queue的消息");
    jmsMessagingTemplate.convertAndSend(topic,"這是Topic的消息");
}

運行后，在控制台只輸出了Queue的消息

問題

Springboot整合ActiveMQ模式只能監聽Queue隊列的消息進行處理，所以如何處理topic消息?

解決：

在Springboot的application.properties文件中添加如下內容

spring.jms.pub-sub-domain=true   //默認是false，開啟發布訂閱模式

啟動測試

經過上述修改，又只能監聽topic的消息，queue的消息又無法獲取。

解決：

只有通過自定義監聽器類來處理

在監聽器類的@JmsListener添加connectionFactory屬性

@Component
public class ActiveMQConsumer {
    //接收queue消息
   @JmsListener(destination = "queue_test",containerFactory =     
                   "queueListenerContainerFactory")
    public void handler(String message){
        System.out.println(message);
    }
    //接收topic消息
    @JmsListener(destination = "topic_test",containerFactory = 
                "topicListenerContainerFactory")
    public void handlerTopic(String msessage){
        System.out.println(msessage);
    }
}

創建一個配置類，在配置類中提供兩個監聽工廠配置

@Configuration
public class ConsumerConfiguration {

    @Value("${spring.activemq.broker-url}")
    private String host;

    @Bean
    public ConnectionFactory getActiveMqConnection(){
        return new ActiveMQConnectionFactory(host);
    }

    @Bean(name="queueListenerContainerFactory")
    public JmsListenerContainerFactory queueListenerContailerFactory(ConnectionFactory connectionFactory){
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setPubSubDomain(false);
        return factory;
    }
    @Bean(name="topicListenerContainerFactory")
    public JmsListenerContainerFactory topicListenerContainerFactory(ConnectionFactory connectionFactory){
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setPubSubDomain(true);
        return factory;
    }
}

運行測試