Springboot整合Active消息隊列

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

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

簡單理解：

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

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

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

<!-- 整合消息隊列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測試，安裝在別的機(jī)器，防火墻和端口號記得開放
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)
#消息隊列默認(rèn)是點(diǎn)對點(diǎn)的，如果需要發(fā)布/訂閱模式那么需要加上下面注解(如果同時需要點(diǎn)對點(diǎn)發(fā)布訂閱這里也需注釋掉)
# spring.jms.pub-sub-domain=true

3、Springboot主類

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

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

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

1、Springboot主類

@SpringBootApplication
@EnableJms
public class Main {

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

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

@Component
public class FirstConsumer {

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

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

@Component
public class TwoConsumer {

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

3、Service類

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

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

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

}

4、ServiceImpl實現(xiàn)類

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

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

5.QueueController類

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

 // 這里后面調(diào)用的是Springboot主類的quene隊列
 @GetMapping("first")
 public Object common(String msg){
  producerService.sendMessage(msg); 
  return "Success";
 }  
 
 // 這個隊列是新建的一個名為two.queue的點(diǎn)對點(diǎn)消息隊列
 @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啟動時候，就生成了這兩個隊列，而且他們都會有一個消費(fèi)者

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

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

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

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

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

Messages Dequeued：消息出隊。

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

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

1.appliaction.properties文件

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

2.Springboot主類添加

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

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

//這里定義了三個消費(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實現(xiàn)類

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

6.Controller類

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

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

7.演示效果：

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

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

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

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

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

效果演示

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

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

2：然而我們在打開topic頁面發(fā)現(xiàn)，這里卻存在一個消費(fèi)者。

所以我個人理解是，當(dāng)同時啟動的時候，所產(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)會默認(rèn)支持quene（點(diǎn)對點(diǎn)模式），但一旦加上這段配置，系統(tǒng)又變成只支持發(fā)布訂閱模式。

那如何同時都可以成功呢？

思路如下：

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

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

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

因為你去掉上面的配置，那么系統(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"看等下是否會打印出
 @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;
  }

效果：