SpringBoot整合ActiveMQ

1.加入依赖

    org.springframework.boot
    spring-boot-starter-activemq



    org.springframework.boot
    spring-boot-starter-web



    org.springframework.boot
    spring-boot-configuration-processor
    true



    org.projectlombok
    lombok
    1.18.4

2.增加配置类

package com.example.mq.config.properties;

import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;

@Configuration
@ConfigurationProperties(prefix = "spring.activemq")
@Data
public class ActiveMqConfigProperties {
    @Value("user")
    private String user;
    @Value("password")
    private String password;
    @Value("broker-url")
    private String brokerUrl;
}

3.配置MQ连接工厂和MQ消费者监听器工厂

package com.example.mq.config;

import com.example.mq.config.properties.ActiveMqConfigProperties;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.config.JmsListenerContainerFactory;

@Configuration
public class ActiveMqConfig {
    @Autowired
    ActiveMqConfigProperties activeMqConfigProperties;

    //创建会话工厂
    @Bean
    public ActiveMQConnectionFactory activeMQConnectionFactory(){
        return new ActiveMQConnectionFactory(activeMqConfigProperties.getUser(),activeMqConfigProperties.getPassword(),activeMqConfigProperties.getBrokerUrl());
    }

    //创建消息监听器-点对点方式
    @Bean
    public JmsListenerContainerFactory jmsListenerContainerFactoryQueue(ActiveMQConnectionFactory activeMQConnectionFactory){
        DefaultJmsListenerContainerFactory defaultJmsListenerContainerFactory = new DefaultJmsListenerContainerFactory();
        defaultJmsListenerContainerFactory.setConnectionFactory(activeMQConnectionFactory);

        return defaultJmsListenerContainerFactory;
    }

    //创建消息监听器-发布/订阅方式
    @Bean
    public JmsListenerContainerFactory jmsListenerContainerFactoryTopic(ActiveMQConnectionFactory activeMQConnectionFactory){
        DefaultJmsListenerContainerFactory defaultJmsListenerContainerFactory = new DefaultJmsListenerContainerFactory();
        defaultJmsListenerContainerFactory.setConnectionFactory(activeMQConnectionFactory);
        defaultJmsListenerContainerFactory.setPubSubDomain(true);//设置发布/订阅方式，默认是点对点方式

        return defaultJmsListenerContainerFactory;
    }
}

4.编写生产者代码

package com.example.mq.controller;

import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.jms.Queue;
import javax.jms.Topic;

@RestController("producer")
public class ProducerController {
    @Autowired
    JmsMessagingTemplate jmsMessagingTemplate;

    @GetMapping("/sendByQueue")
    public String send(){
        //创建点对点队列
        Queue queue = new ActiveMQQueue("testSendQueueMsg");

        for(int i=1;i<=10;i++){
            jmsMessagingTemplate.convertAndSend(queue, i);
        }

        return "点对点消息发送成功";
    }

    @GetMapping("/sendByTopic")
    public String sendTopic(){
        //创建发布/订阅队列
        Topic topic = new ActiveMQTopic("testSendTopicMsg");

        for(int i=1;i<=10;i++){
            jmsMessagingTemplate.convertAndSend(topic,i);
        }

        return "发布/订阅消息发送成功";
    }
}

5.编写消费者监听类

package com.example.mq.listener;

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

@Component
public class ActiveMQListener {

    @JmsListener(destination = "testSendQueueMsg",containerFactory = "jmsListenerContainerFactoryQueue")
    public void receiverQueue(String text){
        System.out.println("点到点收到的消息为=" + text);
    }

    @JmsListener(destination = "testSendTopicMsg",containerFactory = "jmsListenerContainerFactoryTopic")
    public void receiverTopic(String text){
        System.out.println("发布/订阅收到的消息为=" + text);
    }
}

6.点对点方式执行结果

7.发布/订阅方式执行效果