安裝 ActiveMQ(見上一篇)

整合

IDEA 建立 SpringBoot 專案，

1.開啟IDEA,建立新專案，選擇Spring Initializr

2.輸入Artifact

3.勾選Web

4.點選finish完成

5.進入專案，可以將以下內容刪除

因為 SpringBoot 已經內建了對 ActiveMQ 的支援，所以直接引入依賴 spring-boot-starter-activemq 就行。整體專案結構如下：

1、pom.xml 檔案

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>

	<groupId>com.yuezhong</groupId>
	<artifactId>activemq</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<packaging>jar</packaging>

	<name>activemq</name>
	<description>Demo project for Spring Boot ActiveMQ</description>

	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.0.1.RELEASE</version>
		<relativePath/>
	</parent>

	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
		<java.version>1.8</java.version>
	</properties>

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

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

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

	</dependencies>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>
</project>
 

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.send-timeout=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
 

package com.yuezhong.activemq.client;

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

/**
 * Created by yz.shi on 2018/4/9.
 */
@Component
public class ActiveMQClient {

    // 使用JmsListener配置消費者監聽的佇列，其中text是接收到的訊息
    @JmsListener(destination = "mytest.queue")
    public void receiveQueue(String text) {
        if(!StringUtils.isEmpty(text)){
            System.out.println("ActiveMQServer收到的報文為:"+text);
            System.out.println("");
        }
    }
}
 

4、訊息接收類

package com.yuezhong.activemq.server;

import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import javax.jms.Destination;

/**
 * Created by yz.shi on 2018/4/9.
 */
@Component
@EnableScheduling
public class ActiveMQServer {

    @Autowired
    private JmsTemplate jmsTemplate;

    static int index = 0;

    Destination destination = new ActiveMQQueue("mytest.queue");

    @Scheduled(fixedDelay = 5000) // 5s執行一次   只有無參的方法才能用該註解
    public void sendMessage() {
        index++;
        String message = "shiyuezhong" + index;
        this.jmsTemplate.convertAndSend(destination, message);
    }

}

5、ActivemqApplication

package com.yuezhong;

import com.yuezhong.queue.producer.Producer;
import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import javax.annotation.PostConstruct;
import javax.jms.Destination;

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

}

另： Producer

package com.yuezhong.queue.producer;

import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import javax.jms.Destination;
import javax.annotation.Resource;

/**
 * Created by yz.shi on 2018/4/9.
 */
@Service("producer")
public class Producer {
    @Resource
    private JmsTemplate jmsTemplate;

    /**
     * 傳送訊息（主方法設定佇列）
     *
     * @param destination 傳送到的佇列
     * @param message     待發送的訊息
     */
    public void convertAndSend(Destination destination, final String message) {
        jmsTemplate.convertAndSend(destination, message);
    }

    /**
     * 傳送訊息（定時傳送佇列）
     */
    static int i = 100;
    Destination destination = new ActiveMQQueue("yzshi_queue");

    @Scheduled(fixedDelay = 5000) // 5s執行一次   只有無參的方法才能用該註解
    public void convertAndSend() {
        i++;
        String message = "這是定時任務queueProducer傳送的第" + i + "個訊息！";
        jmsTemplate.convertAndSend(destination, message);
    }

}

package com.yuezhong.queue.consumer;

import org.springframework.jms.annotation.JmsListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

/**
 * Created by yz.shi on 2018/4/9.
 */
@Component
public class ConsumerB {
    /**
     * 使用JmsListener配置消費者監聽的佇列
     *
     * @param text 接收到的訊息
     */
    @JmsListener(destination = "yzshi_queue")
    @SendTo("out.queue")
    public String receiveQueue(String text) {
        System.out.println("----------------Consumer-B : 收到的報文為:" + text);
        return text;
    }

    @JmsListener(destination = "yzshi_queue")
    public void printReceiveQueue(String text) {
        System.out.println("+++++++++++++++++Consumer-B : 收到的報文為:" + text);
    }
}

ConsumerC

package com.yuezhong.queue.consumer;

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

/**
 * Created by yz.shi on 2018/4/9.
 */
@Component
public class ConsumerC {
    /**
     * 使用JmsListener配置消費者監聽的佇列
     *
     * @param text 接收到的訊息
     */
    @JmsListener(destination = "out.queue")
    public void consumerMessage(String text) {
        System.out.println("Consumer-C : 從out.queue佇列收到的回覆報文為:" + text);
    }
}

package com.yuezhong;

import com.yuezhong.queue.producer.Producer;
import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import javax.annotation.PostConstruct;
import javax.jms.Destination;

@SpringBootApplication
public class ActivemqApplication {

    @Autowired
    private Producer producer;

    @PostConstruct
    public void init() {
        int num = 10;
        try {
            Destination destinationQueue = new ActiveMQQueue("yzshi_queue");
            for (int i = 1; i <= num; i++) {
                producer.convertAndSend(destinationQueue, "這是主queueProducer傳送的第" + i + "個訊息！");
            }
            System.out.println("activeMQ生產成功！");
        } catch (Exception e) {
            System.out.println("activeMQ生產失敗！");
       }
    } 

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

}