Spring Boot整合Activemq及其使用

什麼是ActiveMQ？

ActiveMQ是一種開源的，實作了JMS規範的，面向消息(MOM)的中間件，為應用程式提供高效的、可擴充的、穩定的和安全的企業級消息通信。

什麼是JMS？

Java消息服務指的是兩個應用程式之間進行異步通信的API，它為标準消息協定和消息服務提供了一組通用接口，包括建立、發送、讀取消息等，用于支援JAVA應用程式開發。在J2EE中，當兩個應用程式使用JMS進行通信時，它們之間并不是直接相連的，而是通過一個共同的消息收發服務連接配接起來，可以達到解耦的效果。

Spring Boot整合ActiveMq

1.添加依賴

2.修改application.properties的配置檔案

spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.in-memory=true           

3.消息生産者代碼

package com.ls.mq;
 
import javax.jms.Destination;
 
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Service;
 
@Service("producer")
public class Producer {
    @Autowired
    private JmsMessagingTemplate jmsTemplate;
    // 發送消息，destination是發送到的隊列，message是待發送的消息
    public void sendMessage(Destination destination, final String message){
        jmsTemplate.convertAndSend(destination, message);
    }
}           

4.消息消費者代碼

package com.ls.mq;
 
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;
 
@Component
public class Consumer {
        // 使用JmsListener配置消費者監聽的隊列，其中text是接收到的消息
    @JmsListener(destination = "mytest.queue")
    public void receiveQueue(String text) {
        System.out.println("Consumer收到的封包為:"+text);
    }
}           

5.進行測試

package com.ls.mq;
 
import javax.jms.Destination;
 
import org.apache.activemq.command.ActiveMQQueue;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
 
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringbootJmsApplicationTests {
    
    @Autowired
    private Producer producer;
    
    @Test
    public void contextLoads() throws InterruptedException {
        Destination destination = new ActiveMQQueue("mytest.queue");
        
        for(int i=0; i<100; i++){
            producer.sendMessage(destination, "lsls!");
        }
    }
 
}           

注意：測試類必須和消息生産者在同層包中，否則将注入失敗

同時發送和接收topic/queue

1.建立一個JMS的配置類

@Configuration
public class JmsConfig {
    public final static String TOPIC = "springboot.topic.test";
    public final static String QUEUE = "springboot.queue.test";

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

   @Bean
    public Queue queue() {
        return new ActiveMQQueue(QUEUE);
    }
    // topic模式的ListenerContainer
    @Bean
    public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ConnectionFactory activeMQConnectionFactory) {
        DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
        bean.setPubSubDomain(true);
        bean.setConnectionFactory(activeMQConnectionFactory);
        return bean;
    }
    // queue模式的ListenerContainer
    @Bean
    public JmsListenerContainerFactory<?> jmsListenerContainerQueue(ConnectionFactory activeMQConnectionFactory) {
        DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
        bean.setConnectionFactory(activeMQConnectionFactory);
        return bean;
    }
}           

2.然後在寫一個消費者類

@Component
public class Consumer2 {
    private final static Logger logger = LoggerFactory.getLogger(JMSConsumer3.class);

    @JmsListener(destination = JmsConfig.TOPIC,containerFactory = "jmsListenerContainerTopic")
    public void onTopicMessage(String msg) {
        logger.info("接收到topic消息：{}",msg);
    }

    @JmsListener(destination = JmsConfig.QUEUE,containerFactory = "jmsListenerContainerQueue")
    public void onQueueMessage(String msg) {
        logger.info("接收到queue消息：{}",text);
    }
}           

3.測試

@Autowired
    private Topic topic;
    @Autowired
    private Queue queue;

    @Test
    public void testJms2() {
        for (int i=0;i<10;i++) {
            jmsProducer.sendMessage(queue,"queue,world!" + i);
            jmsProducer.sendMessage(topic, "topic,world!" + i);
        }
    }