SpringBoot整合ActiveMQ

時間 2020-04-03

原文原文鏈接

ActiveMQ簡介：http://www.javashuo.com/article/p-krkwopju-bm.html
ActiveMQ安裝與使用：http://www.javashuo.com/article/p-blwcetlu-ez.html
MQ理論知識：http://www.javashuo.com/article/p-ypffqfyx-s.html

1.SpringBoot整合ActiveMQ

在SpringBoot中集成ActiveMQ相對仍是比較簡單的，不須要安裝什麼服務，
默認使用內存中的ActiveMQ，配合外置ActiveMQ Server會更好.

1.1 建立工程maven子模塊

1.1.1 建立mq_springBoot子模塊

1.1.2 添加相關依賴

<!-- activemq啓動器    不加這個,用springboot的啓動器也能夠,由於springboot內置了activemq-->
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-activemq</artifactId>
  <version>2.0.5.RELEASE</version>
</dependency>

<!-- springBoot內置,能夠不加,加上會更好
<dependency>
  <groupId>org.apache.activemq</groupId>
  <artifactId>activemq-all</artifactId>
  <version>5.14.0</version>
</dependency>
-->

1.1.3 建立程序入口

package com.manlu;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
 * @author 漫路
 */
@SpringBootApplication
public class MqSbApplication {
    public static void main(String[] args) {
        SpringApplication.run(MqSbApplication.class,args);
    }
}

1.1.4 建立application.yml文件

spring:
  activemq:
    broker-url: tcp://127.0.0.1:61616 #MQ所在的服務器地址
    in-memory: true #是否使用內置的MQ。  true：使用； false：不使用；
    non-blocking-redelivery: false #是否在回滾消息以前中止消息傳遞。當啓用此命令時，消息順序不會被保留
    user: admin # 用戶名
    password: admin # 密碼

配置的全部具體意思以下(properties格式):

spring.activemq.broker-url=tcp://127.0.0.1:61616
# 在考慮結束以前等待的時間
#spring.activemq.close-timeout=15s
# 默認代理URL是否應該在內存中。若是指定了顯式代理，則忽略此值。
spring.activemq.in-memory=true
# 是否在回滾回滾消息以前中止消息傳遞。這意味着當啓用此命令時，消息順序不會被保留。
spring.activemq.non-blocking-redelivery=false
# 密碼
spring.activemq.password=admin
# 等待消息發送響應的時間。設置爲0等待永遠。
spring.activemq.user=admin
# 是否信任全部包
#spring.activemq.packages.trust-all=
# 要信任的特定包的逗號分隔列表（當不信任全部包時）
#spring.activemq.packages.trusted=
# 當鏈接請求和池滿時是否阻塞。設置false會拋「JMSException異常」。
#spring.activemq.pool.block-if-full=true
# 若是池仍然滿，則在拋出異常前阻塞時間。
#spring.activemq.pool.block-if-full-timeout=-1ms
# 是否在啓動時建立鏈接。能夠在啓動時用於加熱池。
#spring.activemq.pool.create-connection-on-startup=true
# 是否用Pooledconnectionfactory代替普通的ConnectionFactory。
#spring.activemq.pool.enabled=false
# 鏈接過時超時。
#spring.activemq.pool.expiry-timeout=0ms
# 鏈接空閒超時
#spring.activemq.pool.idle-timeout=30s
# 鏈接池最大鏈接數
#spring.activemq.pool.max-connections=1
# 每一個鏈接的有效會話的最大數目。
#spring.activemq.pool.maximum-active-session-per-connection=500
# 當有"JMSException"時嘗試從新鏈接
#spring.activemq.pool.reconnect-on-exception=true
# 在空閒鏈接清除線程之間運行的時間。當爲負數時，沒有空閒鏈接驅逐線程運行。
#spring.activemq.pool.time-between-expiration-check=-1ms
# 是否只使用一個MessageProducer
#spring.activemq.pool.use-anonymous-producers=true

1.1.5 建立config

package com.manlu.config;

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

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

/**
 * @author 漫路
 */
@Configuration
public class ActiveMQConfig {
    @Bean
    public Queue queue(){
        return new ActiveMQQueue("manlu.queue");
    }
    
    @Bean
    public Topic topic(){
        return new ActiveMQTopic("manlu.topic");
    }
}

1.2 編寫消費者和生產者

1.2.1 編寫生產者QueueProducer

package com.manlu.mq;

import org.apache.activemq.command.ActiveMQMapMessage;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.jms.MapMessage;
import javax.jms.Queue;

/**
 * 消息的生產者
 *
 * @author 漫路
 */
@Component
@EnableScheduling
public class QueueProducer {
    /*
    * @Resource // 也能夠注入JmsTemplate，JmsMessagingTemplate對JmsTemplate進行了封裝
    * private JmsMessagingTemplate jmsTemplate; //
    * 發送消息，destination是發送到的隊列，message是待發送的消息
    *
    * @Scheduled(fixedDelay=3000)//每3s執行1次
    * public void sendMessage(Destination destination, final String message){
    *   jmsTemplate.convertAndSend(destination, message);
    * }
    */
    @Resource
    private JmsMessagingTemplate jmsMessagingTemplate;

    @Resource
    private Queue queue;

    @Scheduled(fixedDelay = 3000)//3秒執行1次
    public void send() {
        try {
            MapMessage mapMessage = new ActiveMQMapMessage();
            mapMessage.setString("info", "小老弟在敲代碼");
            jmsMessagingTemplate.convertAndSend(queue, mapMessage);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}

1.2.2 編寫消費者QueueConsumer

package com.manlu.mq;

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

import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;

/**
 * 消息的消費者
 * @author 漫路
 */
@Component
public class QueueConsumer {
    //使用JmsListener配置消費者監聽的隊列,其中Message是接收到的消息
    @JmsListener(destination = "manlu.queue")
    public void receiveQueue(Message message){
        try {
            MapMessage mapMessage = (MapMessage) message;
            String info = mapMessage.getString("info");
            System.out.println(info);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}