快捷導(dǎo)航

詳解Springboot整合ActiveMQ（Queue和Topic兩種模式）

更新時(shí)間：2020年04月10日 16:36:40 作者：一枕江風(fēng)

這篇文章主要介紹了詳解Springboot整合ActiveMQ（Queue和Topic兩種模式），文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

寫在前面：從2018年底開始學(xué)習(xí)SpringBoot，也用SpringBoot寫過一些項(xiàng)目。這里對學(xué)習(xí)Springboot的一些知識總結(jié)記錄一下。如果你也在學(xué)習(xí)SpringBoot，可以關(guān)注我，一起學(xué)習(xí)，一起進(jìn)步。

ActiveMQ簡介

1、ActiveMQ簡介

Apache ActiveMQ是Apache軟件基金會所研發(fā)的開放源代碼消息中間件；由于ActiveMQ是一個(gè)純Java程序，因此只需要操作系統(tǒng)支持Java虛擬機(jī)，ActiveMQ便可執(zhí)行。

2、ActiveMQ下載

下載地址：http://activemq.apache.org/components/classic/download/

下載完成后解壓雙擊activemq.bat文件打開(不用安裝，直接使用)，目錄和打開后效果如下：

運(yùn)行后，瀏覽器訪問http://localhost:8161/地址進(jìn)入一下界面。

點(diǎn)擊Manage ActiveMQ broker登錄到ActiveMQ管理頁面，默認(rèn)賬號和密碼都是admin。管理頁面如下：

SpringBoot整合ActiveMQ

1、新建SpringBoot項(xiàng)目

新建Springboot項(xiàng)目，添加對應(yīng)的依賴。項(xiàng)目完整的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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
 <modelVersion>4.0.0</modelVersion>
 <parent>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-parent</artifactId>
  <version>2.2.5.RELEASE</version>
  <relativePath/> <!-- lookup parent from repository -->
 </parent>
 <groupId>com.mcy</groupId>
 <artifactId>springboot-mq</artifactId>
 <version>0.0.1-SNAPSHOT</version>
 <name>springboot-mq</name>
 <description>Demo project for Spring Boot</description>

 <properties>
  <java.version>1.8</java.version>
 </properties>

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

  <!--Activemq依賴-->
  <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-activemq</artifactId>
  </dependency>

  <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-test</artifactId>
   <scope>test</scope>
   <exclusions>
    <exclusion>
     <groupId>org.junit.vintage</groupId>
     <artifactId>junit-vintage-engine</artifactId>
    </exclusion>
   </exclusions>
  </dependency>
 </dependencies>

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

2、項(xiàng)目結(jié)構(gòu)

3、相關(guān)配置信息

在application.properties類中添加ActiveMQ相關(guān)的配置信息

server.port=8080
server.servlet.context-path=/mq

#MQ服務(wù)器地址
spring.activemq.broker-url=tcp://localhost:61616
#用戶名
spring.activemq.user=admin
#密碼
spring.activemq.password=admin
#設(shè)置是Queue隊(duì)列還是Topic，false為Queue，true為Topic，默認(rèn)false-Queue
spring.jms.pub-sub-domain=false
#spring.jms.pub-sub-domain=true

#變量，定義隊(duì)列和topic的名稱
myqueue: activemq-queue
mytopic: activemq-topic

4、ActiveMQ配置類

ActiveMQ配置類ConfigBean，配置了Queue隊(duì)列和topic兩種模式，代碼如下：

import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.stereotype.Component;
import javax.jms.Topic;

/**
 1. MQ配置類
 */
@Component
@EnableJms
public class ConfigBean {
 @Value("${myqueue}")
 private String myQueue;
 @Value("${mytopic}")
 private String topicName;

 //隊(duì)列
 @Bean
 public ActiveMQQueue queue(){
  return new ActiveMQQueue(myQueue);
 }

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

Queue隊(duì)列模式

隊(duì)列模式即點(diǎn)對點(diǎn)傳輸。
點(diǎn)對點(diǎn)消息傳遞域的特點(diǎn)如下：

每個(gè)消息只能有一個(gè)消費(fèi)者，類似于1對1的關(guān)系。好比個(gè)人快遞自己領(lǐng)自己的。

消息的生產(chǎn)者和消費(fèi)者之間沒有時(shí)間上的相關(guān)性。無論消費(fèi)者在生產(chǎn)者發(fā)送消息的時(shí)候是否處于運(yùn)行狀態(tài)，消費(fèi)者都可以提取消息。好比我們的發(fā)送短信，發(fā)送者發(fā)送后不見得接收者會即收即看。

消息被消費(fèi)后隊(duì)列中不會再存儲，所以消費(fèi)者不會消費(fèi)到已經(jīng)被消費(fèi)掉的消息。

1、隊(duì)列生產(chǎn)者

QueueProducerController類為隊(duì)列生產(chǎn)者控制器，主要向消息隊(duì)列中發(fā)送消息。代碼如下：

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;
import javax.jms.Queue;

/*
 * 隊(duì)列消息生產(chǎn)者
 */
@RestController
public class QueueProducerController {
 @Autowired
 private JmsMessagingTemplate jmsMessagingTemplate;

 @Autowired
 private Queue queue;

 /*
  * 消息生產(chǎn)者
  */
 @RequestMapping("/sendmsg")
 public void sendmsg(String msg) {
  System.out.println("發(fā)送消息到隊(duì)列：" + msg);
  // 指定消息發(fā)送的目的地及內(nèi)容
  this.jmsMessagingTemplate.convertAndSend(this.queue, msg);
 }
}

2、隊(duì)列消費(fèi)者

QueueConsumerController類為隊(duì)列消費(fèi)者控制器，具體代碼如下：

import org.springframework.beans.factory.annotation.Value;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.web.bind.annotation.RestController;

/*
 1. 隊(duì)列queue消費(fèi)者控制器
 */
@RestController
public class QueueConsumerController {
 /*
  * 消費(fèi)者接收消息
  */
 @JmsListener(destination="${myqueue}")
 public void readActiveQueue(String message) {
  System.out.println("接受到：" + message);
 }
}

3、測試效果

運(yùn)行項(xiàng)目在瀏覽器中訪問http://localhost:8080/mq/sendmsg?msg=123。向消息隊(duì)列中發(fā)送123。控制臺輸出效果：

ActiveMQ控制臺顯示：

Number Of Pending Messages：消息隊(duì)列中待處理的消息
Number Of Consumers：消費(fèi)者的數(shù)量
Messages Enqueued：累計(jì)進(jìn)入過消息隊(duì)列的總量
Messages Dequeued：累計(jì)消費(fèi)過的消息總量

【注】隊(duì)列模式時(shí)，配置文件application.properties中spring.jms.pub-sub-domain屬性必須設(shè)置為false。

Topic模式

topic模式基于發(fā)布/訂閱模式的傳輸。
基于發(fā)布/訂閱模式的傳輸?shù)奶攸c(diǎn)如下：

生產(chǎn)者將消息發(fā)布到topic中，每個(gè)消息可以有多個(gè)消費(fèi)者，屬于1：N的關(guān)系；
生產(chǎn)者和消費(fèi)者之間有時(shí)間上的相關(guān)性。訂閱某一個(gè)主題的消費(fèi)者只能消費(fèi)自它訂閱之后發(fā)布的消息。
生產(chǎn)者生產(chǎn)時(shí)，topic不保存消息它是無狀態(tài)的不落地，假如無人訂閱就去生產(chǎn)，那就是一條廢消息。

1、topic生產(chǎn)者

TopicProducerController類為topic生產(chǎn)者控制器，主要向消息隊(duì)列中發(fā)送消息。代碼如下：

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;
import javax.jms.Queue;
import javax.jms.Topic;

/*
* topic消息生產(chǎn)者
*/
@RestController
public class TopicProducerController {
 @Autowired
 private JmsMessagingTemplate jmsMessagingTemplate;

 @Autowired
 private Topic topic;

 /*
 * 消息生產(chǎn)者
 */
 @RequestMapping("/topicsendmsg")
 public void sendmsg(String msg) {
  System.out.println("發(fā)送消息到MQ：" + msg);
  // 指定消息發(fā)送的目的地及內(nèi)容
  this.jmsMessagingTemplate.convertAndSend(this.topic, msg);
 }
}

2、topic消費(fèi)者

TopicConsumerController類為topic消費(fèi)者控制器，其中寫了兩個(gè)消費(fèi)者方法，可以理解為有兩個(gè)用戶訂閱。具體代碼如下：

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

/*
 1. topic消費(fèi)者控制器
 */
@RestController
public class TopicConsumerController {
 /*
  * 消費(fèi)者接收消息
  */
 @JmsListener(destination="${mytopic}")
 public void readActiveQueue(String message) {
  System.out.println("接受到：" + message);
 }

 @JmsListener(destination="${mytopic}")
 public void readActiveQueue1(String message) {
  System.out.println("接受到：" + message);
 }
}

3、測試效果

運(yùn)行項(xiàng)目在瀏覽器中訪問http://localhost:8080/mq/topicsendmsg?msg=123。向消息隊(duì)列中發(fā)送123?？刂婆_輸出效果（有兩個(gè)消費(fèi)者方法）：