springbooot整合ActiveMQ实现消息队列和主题

一、Springbot整合ActiveMQ需要引入依赖

1.pom

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

2.application.yml

# activemq的配置
spring:
  activemq:
    broker-url: tcp://127.0.0.1:61616
    user: admin
    password: admin
    pool:
      max-connections: 100
  jms:
    pub-sub-domain: false # false代表队列，true代表topic

# 队列的名称 常量 @Value(${})获取属性
myqueue: boot-active-queue
mytopic: boot-active-topic      

二、整合Queue

1.QueueConfig.java

@Component
public class QueueConfig {

    @Value("${myqueue}")
    private String myQueue;

    @Bean
    public Queue queue(){
        return new ActiveMQQueue(myQueue);
    }
}      

2.QueueProducer.java

@Component
public class QueueProducer {

    @Autowired
    private JmsMessagingTemplate jmsMessagingTemplate;

    @Autowired
    private Queue queue;

    // 每隔3s发送
    @Scheduled(fixedDelay = 3000)
    public void produceMsg(){
        jmsMessagingTemplate.convertAndSend(queue,"message"+ new Date().toLocaleString());
    }
}      

3.QueueConsumer.java

@Component
public class QueueConsumer {

    //监测消息，立马消费
    @JmsListener(destination = "${myqueue}")
    public void receive(TextMessage textMessage) throws JMSException {
        System.out.println("消费到的消息是："+ textMessage.getText());
    }
}      

三、整合Topic主题

1.TopicConfig.java

@Component
public class TopicConfig {

    @Value("${mytopic}")
    private String myTopic;

    @Bean
    public ActiveMQTopic topic(){
        return new ActiveMQTopic(myTopic);
    }
}      

2.TopicProducer.java

@Component
public class TopicProducer {

    @Autowired
    private JmsMessagingTemplate jmsMessagingTemplate;

    @Autowired
    private Topic topic;

    // 每隔3s发送
    @Scheduled(fixedDelay = 3000)
    public void produceMsg(){
        //设置成主题 方式
        jmsMessagingTemplate.getJmsTemplate().setPubSubDomain(true);
        jmsMessagingTemplate.convertAndSend(topic,"message"+ new Date().toLocaleString());
        System.out.println("发送消息。。。");
    }
}      

3.TopicConsumer.java

@Component
public class TopicConsumer {

    //监测消息，立马消费
    @JmsListener(destination = "${mytopic}")
    public void receive(TextMessage textMessage) throws JMSException {
        System.out.println("消费到的主题消息是："+ textMessage.getText());
    }
}      

四、测试启动

@SpringBootApplication
@EnableJms //启用消息队列
@EnableScheduling //定时任务
public class SpringbootAvtivemqApplication {

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

}