Springboot中怎么對(duì)ActiveMQ進(jìn)行整合

Springboot中怎么對(duì)ActiveMQ進(jìn)行整合，很多新手對(duì)此不是很清楚，為了幫助大家解決這個(gè)難題，下面小編將為大家詳細(xì)講解，有這方面需求的人可以來學(xué)習(xí)下，希望你能有所收獲。

1、首先新建一個(gè)springboot工程（新建過程略）,本文springboot版本 是2.1.1.RELEASE

2、在pom.xml文件添加activemq的相關(guān)依賴

<!--activemq消息隊(duì)列-->

              <dependency>

                  <groupId>org.springframework.boot</groupId>

                  <artifactId>spring-boot-starter-activemq</artifactId>

              </dependency>

              <!--消息隊(duì)列連接池  此處使用2.0+的版本-->

              <dependency>

                  <groupId>org.apache.activemq</groupId>

                  <artifactId>activemq-pool</artifactId>

                 <!--  <version>5.15.0</version> -->

              </dependency>

              <!-- 消息隊(duì)列連接池  此處使用2.1+的版本 -->

              <dependency>

                  <groupId>org.messaginghub</groupId>

                  <artifactId>pooled-jms</artifactId>

              </dependency>

其中連接池相關(guān)的依賴可以不用配置

3、配置application.properties 或者application.yml 本文以application.properties為例

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

spring.activemq.user=admin

spring.activemq.password=admin

#默認(rèn)情況下activemq提供的是queue模式，若要使用topic模式需要配置下面配置

#spring.jms.pub-sub-domain=true

#true 表示使用內(nèi)置的MQ，false則連接服務(wù)器

spring.activemq.in-memory=false

#true表示使用連接池；false時(shí)，每發(fā)送一條數(shù)據(jù)創(chuàng)建一個(gè)連接

spring.activemq.pool.enabled=true

#連接池最大連接數(shù)

spring.activemq.pool.max-connections=10

#空閑的連接過期時(shí)間，默認(rèn)為30秒

spring.activemq.pool.idle-timeout=15000

spring.activemq.pool.enabled=true時(shí)要在pom文件中添加連接池pool相關(guān)的依賴，為false時(shí)不用添加連接池pool相關(guān)的依賴；

若使用連接池pool配置時(shí)，注意兩種依賴的配置否則啟動(dòng)失敗。

工程結(jié)構(gòu)如下圖

Demo代碼如下

package com.example.acmpp.config;

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

import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class BeanConfig {
	
	//定義存放消息的隊(duì)列
    @Bean
    public Queue queue() {
        return new ActiveMQQueue("ActiveMQQueue");
    }
    //定義存放消息的隊(duì)列
    @Bean
    public Topic topic() {
        return new ActiveMQTopic("ActiveMQTopic");
    }
}

消息生產(chǎn)者代碼

import javax.jms.Queue;

import javax.jms.Topic;



import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.jms.core.JmsMessagingTemplate;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RestController;

/*

 *

 * 消息生產(chǎn)者

 */

@RestController

public class ProviderController {

      

       @Autowired

       private JmsMessagingTemplate jmsMessagingTemplate;



       @Autowired

       private Queue queue;



       @Autowired

       private Topic topic;



       /**

         * 消息生產(chǎn)者 Queue模式

        *

        */

       @RequestMapping("/sendQ")

       public void sendQ(String msg) {

              //方法一：添加消息到消息隊(duì)列

        jmsMessagingTemplate.convertAndSend(queue, msg);

        //方法二：這種方式不需要手動(dòng)創(chuàng)建queue，系統(tǒng)會(huì)自行創(chuàng)建名為test的隊(duì)列

        //jmsMessagingTemplate.convertAndSend("testQ", msg);

       }

       /**

        * 消息生產(chǎn)者 Topic模式

        * @param msg

        */

       @RequestMapping("/sendT")

       public void sendT(String msg) {

              // 指定消息發(fā)送的目的地及內(nèi)容

              System.out.println("@@@@@@@@@@@@@@" + msg);

              this.jmsMessagingTemplate.convertAndSend(this.topic, msg);

       }
}

消息消費(fèi)者代碼

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.jms.annotation.JmsListener;

import org.springframework.jms.core.JmsMessagingTemplate;

import org.springframework.messaging.handler.annotation.SendTo;

import org.springframework.stereotype.Component;



/**

 * 消息消費(fèi)者

 * @author FFF

 *

 */

@Component

public class ConsumerService {

       @Autowired

    private JmsMessagingTemplate jmsMessagingTemplate;

      

       /**

        * 消費(fèi)ActiveMQQueue

        */

    // 使用JmsListener配置消費(fèi)者監(jiān)聽的隊(duì)列，其中name是接收到的消息

    @JmsListener(destination = "ActiveMQQueue")

    // SendTo 會(huì)將此方法返回的數(shù)據(jù), 寫入到 OutQueue 中去.

    @SendTo("SQueue")

    public String handleMessage(String name) {

        System.out.println("ActiveMQQueue成功接受Name" + name);

        return "ActiveMQQueue成功接受Name" + name;

    }

    /**

        * 消費(fèi)ActiveMQ.DLQ

        */

    // 使用JmsListener配置消費(fèi)者監(jiān)聽的隊(duì)列，其中name是接收到的消息

    @JmsListener(destination = "ActiveMQ.DLQ")

    public void DLQ(String name) {

        System.out.println("ActiveMQ.DLQ成功接受Name==" + name);

    }

    /**

        * 消費(fèi)SQueue

        */

    // 使用JmsListener配置消費(fèi)者監(jiān)聽的隊(duì)列，其中name是接收到的消息

    @JmsListener(destination = "SQueue")

    public void SQueue(String name) {

        System.out.println("SQueue成功接受Name==" + name);

    }

    /**

        * 消費(fèi)testQ

        */

    // 使用JmsListener配置消費(fèi)者監(jiān)聽的隊(duì)列，其中name是接收到的消息

    @JmsListener(destination = "testQ")

    public void testQMessage(String name) {

        System.out.println("testQ成功接受Name" + name);

    }

    /**

        * 消費(fèi)topic

        *

        */

    // 使用JmsListener配置消費(fèi)者監(jiān)聽的隊(duì)列，其中name是接收到的消息

    @JmsListener(destination = "ActiveMQTopic")

    public void topicMessage(String name) {

        System.out.println("topicMessage成功接受Name" + name);

    }
}

看完上述內(nèi)容是否對(duì)您有幫助呢？如果還想對(duì)相關(guān)知識(shí)有進(jìn)一步的了解或閱讀更多相關(guān)文章，請(qǐng)關(guān)注億速云行業(yè)資訊頻道，感謝您對(duì)億速云的支持。