ActiveMQ與SpringBoot消息接收方式

同步接收

jmsTemplate.receive()

一個(gè)線程在工作，沒接收到就等待，接收到了就往下執(zhí)行直到程序結(jié)束;

如果想循環(huán)不斷地接收，那么就寫個(gè)while true循環(huán)。

異步接收

使用監(jiān)聽器監(jiān)聽ActiveMQ目的地，當(dāng)有消息的時(shí)候，回調(diào)onMessage方法對消息進(jìn)行處理，ActiveMQ與SpringBoot集成異步接收消息有兩種實(shí)現(xiàn)方式。

1、@JmsListener注解方式實(shí)現(xiàn)

A、 創(chuàng)建SpringBoot工程13-activemq-boot-receiver-async-01

B、 創(chuàng)建的工程的時(shí)候，勾選集成ActiveMQ，會(huì)在pom.xml中添加如下依賴

<!--SpringBoot集成ActiveMQ的起步依賴-->
<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

C、 在application.properties文件中配置ActiveMQ信息

#配置activemq的連接信息
spring.activemq.broker-url=tcp://192.168.235.128:61616
#目的地
spring.jms.template.default-destination=bootQueue
#默認(rèn)是緩存了jms的session的，所以主程序發(fā)送完消息后，不會(huì)退出
# 改為false主程序才可以退出 從SpringBoot2.1.0以后新增的
spring.jms.cache.enabled=false

D、 在com.bjpowernode.activemq.service包下創(chuàng)建MessageService類接收消息

@Service
public class MessageService {
    /**
     * SpringBoot程序啟動(dòng)后，會(huì)掃描到當(dāng)前service類
     * 監(jiān)聽器對象會(huì)被創(chuàng)建
     * 通過注解指定了該監(jiān)聽器對象監(jiān)聽的目的地
     * 如果目的地上有消息需要接收，會(huì)調(diào)用我們自定義的方法，并傳遞接收到的消息到方法中
     * 方法名可以任意
     */
    @JmsListener(destination="${spring.jms.template.default-destination}")
    public void receiveMessage(Message message){
        if(message instanceof TextMessage){
            try {
                String text = ((TextMessage) message).getText();
                System.out.println("SpringBoot異步接收到的消息為：" + text);
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}

E、 運(yùn)行Application程序，發(fā)送一個(gè)消息，查看接收效果

2、@Configuration配置方式實(shí)現(xiàn)

這種方式其實(shí)就是將spring接收異步請求需要配置的xml文件信息，放到配置類中。

A、 創(chuàng)建SpringBoot工程13-activemq-boot-receiver-async-02

B、 創(chuàng)建的工程的時(shí)候，勾選集成ActiveMQ，會(huì)在pom.xml中添加如下依賴

<!--SpringBoot集成ActiveMQ的起步依賴-->
<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

C、 在application.properties文件中配置ActiveMQ信息

#配置activemq的連接信息
spring.activemq.broker-url=tcp://192.168.235.128:61616
#目的地
spring.jms.template.default-destination=bootQueue
#默認(rèn)是緩存了jms的session的，所以主程序發(fā)送完消息后，不會(huì)退出
# 改為false主程序才可以退出 從SpringBoot2.1.0以后新增的
spring.jms.cache.enabled=false

D、 自定義監(jiān)聽器接收處理消息(可以從12-activemq-spring-receiver-async拷貝)

該監(jiān)聽器沒有在配置文件中注冊bean，所有需要加@Component讓Spring容器掃描

E、 在com.bjpowernode.activemq.config包下創(chuàng)建ActiveMQConfig類，用于替換我們前面在applicationContex-jms中配置的信息

@Configuration//相當(dāng)于applicationContext-jms.xml文件
public class ActiveMQConfig {
    /*
        <!-- 配置一個(gè)連接工廠 -->
         <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
         <property name="brokerURL" value="tcp://192.168.235.128:61616"/>
         </bean>
        因?yàn)镾pringBoot已經(jīng)自動(dòng)配置好了ActiveMQ的連接工廠，所有我們在配置文件中就不需要對這個(gè)bean進(jìn)行配置
        我們在使用的時(shí)候，直接注入即可
     */
    @Autowired
    private ActiveMQConnectionFactory connectionFactory;

    @Autowired //注入我們自定義的監(jiān)聽器
    private MyMessageListener myMessageListener;

    //通過@Value注解接收SpringBoot配置文件中的目的地配置
    @Value("${spring.jms.template.default-destination}")
    private String destination;

    /**
         <!-- 配置一個(gè)sping監(jiān)聽器的容器 -->
         <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
         <!--引用連接工廠-->
         <property name="connectionFactory" ref="connectionFactory"/>
         <!--指定監(jiān)聽的目的地-->
         <property name="destinationName" value="springQueue"/>
         <!--監(jiān)聽到消息后，會(huì)回調(diào)onMessage方法，我們在自定義的監(jiān)聽器中對onMessage方法進(jìn)行重寫，完成消息的接收-->
         <property name="messageListener" ref="myMessageListener" />

         <!--pubSubDomain=false表示點(diǎn)對點(diǎn)消息，true表示發(fā)布訂閱消息，默認(rèn)是false-->
         <property name="pubSubDomain" value="true"/>
         </bean>
     */
    @Bean //@Bean注解就相當(dāng)于配置文件的bean標(biāo)簽
    public DefaultMessageListenerContainer defaultMessageListenerContainer(){
        DefaultMessageListenerContainer listenerContainer = new DefaultMessageListenerContainer();
        listenerContainer.setConnectionFactory(connectionFactory);
        listenerContainer.setDestinationName(destination);
        listenerContainer.setMessageListener(myMessageListener);
        //目前我們默認(rèn)就是點(diǎn)對點(diǎn)的消息，暫時(shí)不配置發(fā)布訂閱
        return listenerContainer;
    }
}

 F、 運(yùn)行Application程序，發(fā)送一個(gè)消息，查看接收效果