Springboot整合Active消息隊(duì)列

更新時(shí)間：2020年12月04日 15:12:04 作者：雨點(diǎn)的名字

這篇文章主要介紹了Springboot整合Active消息隊(duì)列的步驟，幫助大家更好的理解和使用springboot框架，感興趣的朋友可以了解下

簡單理解：

Active是Apache公司旗下的一個(gè)消息總線，ActiveMQ是一個(gè)開源兼容Java Message Service(JMS) 面向消息的中件間. 是一個(gè)提供松耦合的應(yīng)用程序架構(gòu).

主要用來在服務(wù)與服務(wù)之間進(jìn)行異步通信的。

一、搭建步驟
1、相應(yīng)jar包

<!-- 整合消息隊(duì)列ActiveMQ -->
  <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-activemq</artifactId>
  </dependency>
  
 <!-- 如果配置線程池則加入 -->
  <dependency> 
   <groupId>org.apache.activemq</groupId> 
   <artifactId>activemq-pool</artifactId> 
  </dependency>

2、application.properties文件

#整合jms測(cè)試，安裝在別的機(jī)器，防火墻和端口號(hào)記得開放
spring.activemq.broker-url=tcp://47.96.44.110:61616

spring.activemq.user=admin
spring.activemq.password=admin
#下列配置要增加依賴
spring.activemq.pool.enabled=true
spring.activemq.pool.max-connections=100

#集群配置(后續(xù)需要在配上)
#spring.activemq.broker-url=failover:(tcp://localhost:61616,tcp://localhost:61617)
#消息隊(duì)列默認(rèn)是點(diǎn)對(duì)點(diǎn)的，如果需要發(fā)布/訂閱模式那么需要加上下面注解(如果同時(shí)需要點(diǎn)對(duì)點(diǎn)發(fā)布訂閱這里也需注釋掉)
# spring.jms.pub-sub-domain=true

3、Springboot主類

<!-- 主類需要多加一個(gè)@EnableJms注解，不過貌似我沒有加的時(shí)候，也能運(yùn)行，為安全起見姑且加上 -->
@SpringBootApplication
@EnableJms

4.5.......根據(jù)不同消息模式來寫了。

二、點(diǎn)對(duì)點(diǎn)案例
我在這里案例中創(chuàng)建了兩個(gè)點(diǎn)對(duì)點(diǎn)隊(duì)列，所以他會(huì)有兩個(gè)queue對(duì)象，同樣對(duì)應(yīng)每個(gè)queue對(duì)象，都會(huì)有單一對(duì)應(yīng)的消費(fèi)者。

1、Springboot主類

@SpringBootApplication
@EnableJms
public class Main {

 public static void main(String[] args) {
  SpringApplication.run(Main.class, args);
 }
 
 //新建一個(gè)的Queue對(duì)象，交給sringboot管理，這個(gè)queue的名稱叫"first.queue".
 @Bean
 public Queue queue(){
  return new ActiveMQQueue("first.queue");
 }
}

2.1、first.queue對(duì)應(yīng)消費(fèi)者

@Component
public class FirstConsumer {

 //名為"first.queue"消息隊(duì)列的消費(fèi)者，通過JmsListener進(jìn)行監(jiān)聽有沒有消息，有消息會(huì)立刻讀取過來
 @JmsListener(destination="first.queue")
 public void receiveQueue(String text){
  System.out.println("FirstConsumer收到的報(bào)文為:"+text);
 }
}

2.2、two.queue對(duì)應(yīng)消費(fèi)者（后面會(huì)創(chuàng)建）

@Component
public class TwoConsumer {

 //名為"two.queue"消息隊(duì)列的消費(fèi)者
 @JmsListener(destination="two.queue")
 public void receiveQueue(String text){
  System.out.println("TwoConsumer收到的報(bào)文為:"+text);
 }
}

3、Service類

/**
 * 功能描述：消息生產(chǎn)
 */
public interface ProducerService {

 // 功能描述：指定消息隊(duì)列，還有消息 
 public void sendMessage(Destination destination, final String message);
 

 // 功能描述：使用默認(rèn)消息隊(duì)列， 發(fā)送消息
 public void sendMessage( final String message);

}

4、ServiceImpl實(shí)現(xiàn)類

/**
 * 功能描述：消息生產(chǎn)者實(shí)現(xiàn)類
 */
@Service
public class ProducerServiceImpl implements ProducerService{

 //這個(gè)隊(duì)列就是Springboot主類中bean的對(duì)象
 @Autowired
 private Queue queue;
 
 //用來發(fā)送消息到broker的對(duì)象,可以理解連接數(shù)據(jù)庫的JDBC
 @Autowired
 private JmsMessagingTemplate jmsTemplate; 
 
 //發(fā)送消息，destination是發(fā)送到的隊(duì)列，message是待發(fā)送的消息
 @Override
 public void sendMessage(Destination destination, String message) {  
  jmsTemplate.convertAndSend(destination, message); 
 }
 
 //發(fā)送消息，queue是發(fā)送到的隊(duì)列，message是待發(fā)送的消息
 @Override
 public void sendMessage(final String message) { 
  jmsTemplate.convertAndSend(this.queue, message); 
 }  
}

5.QueueController類

/**
 * 功能描述：點(diǎn)對(duì)點(diǎn)消息隊(duì)列控制層
 */
@RestController
@RequestMapping("/api/v1")
public class QueueController {
 
 @Autowired
 private ProducerService producerService;  

 // 這里后面調(diào)用的是Springboot主類的quene隊(duì)列
 @GetMapping("first")
 public Object common(String msg){
  producerService.sendMessage(msg); 
  return "Success";
 }  
 
 // 這個(gè)隊(duì)列是新建的一個(gè)名為two.queue的點(diǎn)對(duì)點(diǎn)消息隊(duì)列
 @GetMapping("two")
 public Object order(String msg){
  
  Destination destination = new ActiveMQQueue("two.queue");
  producerService.sendMessage(destination, msg);
  
  return "Success";
 }  
}

6、案例演示：

從演示效果可以得出以下結(jié)論：

1：當(dāng)springboot啟動(dòng)時(shí)候，就生成了這兩個(gè)隊(duì)列，而且他們都會(huì)有一個(gè)消費(fèi)者

2：當(dāng)我通過頁面訪問的時(shí)候，就相當(dāng)于生產(chǎn)者把消息放到隊(duì)列中，一旦放進(jìn)去就會(huì)被消費(fèi)者監(jiān)聽到，就可以獲取生產(chǎn)者放進(jìn)去的值并在后臺(tái)打印出

順便對(duì)頁面中四個(gè)單詞進(jìn)行解釋：

Number Of Pending Messages ：待處理消息的數(shù)量。我們每次都會(huì)被監(jiān)聽處理掉，所以不存在待處理，如果存在就說這里面哪里出故障了，需要排查

Number Of Consumers ：消費(fèi)者數(shù)量

Messages Enqueued：消息排列，這個(gè)只增不見，代表已經(jīng)處理多少消息

Messages Dequeued：消息出隊(duì)。

三、發(fā)布/訂閱者模式

在上面點(diǎn)對(duì)點(diǎn)代碼的基礎(chǔ)上，添加發(fā)布/訂閱相關(guān)代碼

1.appliaction.properties文件

#消息隊(duì)列默認(rèn)是點(diǎn)對(duì)點(diǎn)的，如果需要發(fā)布/訂閱模式那么需要加上下面注解(如果同時(shí)需要點(diǎn)對(duì)點(diǎn)發(fā)布訂閱這里也需注釋掉)
spring.jms.pub-sub-domain=true

2.Springboot主類添加

//新建一個(gè)topic隊(duì)列
 @Bean
 public Topic topic(){
  return new ActiveMQTopic("video.topic");
 }

3.添加多個(gè)消費(fèi)者類

//這里定義了三個(gè)消費(fèi)者
@Component
public class TopicSub {
 
 @JmsListener(destination="video.topic")
 public void receive1(String text){
  System.out.println("video.topic 消費(fèi)者:receive1="+text);
 }
  
 @JmsListener(destination="video.topic")
 public void receive2(String text){
  System.out.println("video.topic 消費(fèi)者:receive2="+text);
 }
  
 @JmsListener(destination="video.topic")
 public void receive3(String text){
  System.out.println("video.topic 消費(fèi)者:receive3="+text);
 } 
}

4.Service類

 //功能描述：消息發(fā)布者
 public void publish(String msg);

5.ServiceImpl實(shí)現(xiàn)類

//=======發(fā)布訂閱相關(guān)代碼=========
 
  @Autowired
  private Topic topic;
    
   @Override
  public void publish(String msg) {
   this.jmsTemplate.convertAndSend(this.topic, msg);
   
  }

6.Controller類

// 這個(gè)隊(duì)列是新建的一個(gè)名為two.queue的點(diǎn)對(duì)點(diǎn)消息隊(duì)列
  @GetMapping("topic")
  public Object topic(String msg){

   producerService.publish(msg);
   
   return "Success";
  }

7.演示效果：

從演示效果總結(jié)如下：

1：Springboot啟動(dòng)的時(shí)候，在Topics目錄下，一共出現(xiàn)了5個(gè)消費(fèi)者。first.queue一個(gè)消費(fèi)者、two.queue一個(gè)消費(fèi)者、video.topic三個(gè)消費(fèi)者

2：當(dāng)我在控制臺(tái)輸入信息后，video.topic的三個(gè)消費(fèi)者都會(huì)監(jiān)聽video.topic發(fā)布的消息，并在控制臺(tái)打印。

四、如何讓點(diǎn)對(duì)點(diǎn)和發(fā)布訂閱同時(shí)有效

為什么這么說呢，因?yàn)楫?dāng)我向上面一樣同時(shí)開啟，會(huì)發(fā)現(xiàn)點(diǎn)對(duì)點(diǎn)模式已經(jīng)失效了。

效果演示

從演示效果，可以得出如下結(jié)論：

1：我們發(fā)現(xiàn)我們?cè)陧撁孑斎?.../two?msg=555消息后，后臺(tái)并沒有成功打印消息。再看Active界面發(fā)現(xiàn)，這個(gè)queue對(duì)象，確實(shí)有一條待處理的消息，但是我們發(fā)現(xiàn)，它對(duì)應(yīng)的消費(fèi)者數(shù)量是為0.

2：然而我們?cè)诖蜷_topic頁面發(fā)現(xiàn)，這里卻存在一個(gè)消費(fèi)者。

所以我個(gè)人理解是，當(dāng)同時(shí)啟動(dòng)的時(shí)候，所產(chǎn)生的消費(fèi)者默認(rèn)都是Topic消費(fèi)者，沒有Queue消費(fèi)者，所以它監(jiān)聽不到queue所待處理的消息。

當(dāng)配置文件不加：spring.jms.pub-sub-domain=true 那么系統(tǒng)會(huì)默認(rèn)支持quene（點(diǎn)對(duì)點(diǎn)模式），但一旦加上這段配置，系統(tǒng)又變成只支持發(fā)布訂閱模式。

那如何同時(shí)都可以成功呢？

思路如下：

第一步：還是需要去掉配置文件中的：

#消息隊(duì)列默認(rèn)是點(diǎn)對(duì)點(diǎn)的，如果需要發(fā)布/訂閱模式那么需要加上下面注解(如果同時(shí)需要點(diǎn)對(duì)點(diǎn)發(fā)布訂閱這里也需注釋掉)
#spring.jms.pub-sub-domain=true

第二步：在發(fā)布訂閱者的中消費(fèi)者中指定獨(dú)立的containerFactory

因?yàn)槟闳サ羯厦娴呐渲茫敲聪到y(tǒng)就默認(rèn)是queue，所以@JmsListener如果不指定獨(dú)立的containerFactory的話是只能消費(fèi)queue消息

@JmsListener(destination="video.topic", containerFactory="jmsListenerContainerTopic")
 public void receive1(String text){
  System.out.println("video.topic 消費(fèi)者:receive1="+text);
 }
 
 
 @JmsListener(destination="video.topic", containerFactory="jmsListenerContainerTopic")
 public void receive2(String text){
  System.out.println("video.topic 消費(fèi)者:receive2="+text);
 }
 
 //第三步我不添加containerFactory="jmsListenerContainerTopic"看等下是否會(huì)打印出
 @JmsListener(destination="video.topic")
 public void receive3(String text){
  System.out.println("video.topic 消費(fèi)者:receive3="+text);
 }

第三步：定義獨(dú)立的topic定義獨(dú)立的JmsListenerContainer

在springboot主類中添加：

@Bean
  public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ConnectionFactory activeMQConnectionFactory) {
   DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
   bean.setPubSubDomain(true);
   bean.setConnectionFactory(activeMQConnectionFactory);
   return bean;
  }

效果：