1. 程式人生 > 程式設計 >Spring Boot ActiveMQ釋出/訂閱訊息模式原理解析

Spring Boot ActiveMQ釋出/訂閱訊息模式原理解析

本文在《Spring Boot基於Active MQ實現整合JMS》的基礎上,介紹如何使用ActiveMQ的釋出/訂閱訊息模式。釋出/訂閱訊息模式是訊息傳送者傳送訊息到主題(topic),而多個訊息接收者監聽這個主題;其中,訊息傳送者和接收者分別叫做釋出者(publisher)和訂閱者(subscriber),對於釋出者來說,它和所有的訂閱者就構成了一個1對多的關係。這種關係如下圖所示:

Spring Boot ActiveMQ釋出/訂閱訊息模式原理解析

釋出/訂閱模式的工作示意圖

訊息生產者將訊息(釋出)到topic中,可以同時有多個訊息消費者(訂閱)消費該訊息。

和點對點方式不同,釋出到topic的訊息會被所有訂閱者消費;當生產者釋出訊息時,不管是否有消費者,都不會儲存訊息;一定要先有訊息的消費者,後有訊息的生產者。

軟體環境

  • ActiveMQ 5.15.13
  • java version 13.0.1
  • IntelliJ IDEA 2019.3.2 (Ultimate Edition)
  • Spring Boot 2.3.0.RELEASE

配置ActiveMQ連線資訊

spring.activemq.broker-url=tcp://127.0.0.1:61616
spring.activemq.in-memory=true
spring.activemq.pool.enabled=false
spring.activemq.password=admin
spring.activemq.user=admin
#預設值false,表示point to point(點到點)模式,true時代表釋出訂閱模式,需要手動開啟
#spring.jms.pub-sub-domain=true

建立生產者和消費者

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Service;

import javax.jms.Destination;

/**
 * 生產者
 */
@Service
public class Publisher {
  @Autowired
  private JmsMessagingTemplate jmsMsgTemplate;

  /**
   * 傳送topic
   *
   * @param destination
   * @param message
   */
  public void publish(Destination destination,String message) {
    jmsMsgTemplate.convertAndSend(destination,message);
  }
}
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Service;

/**
 * 消費者
 */
@Service
public class Subscriber2 {
  private static Logger logger = LoggerFactory.getLogger(Subscriber2.class);

  @JmsListener(destination = "topicListener2")
  public void subscriber(String text) {
    logger.info("Subscriber2 收到的報文:{}",text);
  }
}
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

import javax.jms.JMSException;

/**
 * 消費者
 */
@Component
public class Subscriber1 {
  private static Logger logger = LoggerFactory.getLogger(Subscriber1.class);

  /**
   * 訂閱 topicListener1
   *
   * @param text
   * @throws JMSException
   */
  @JmsListener(destination = "topicListener1")
  public void subscriber(String text) {
    logger.info("Subscriber1 收到的報文:{}",text);
  }

}

釋出訂閱模式和點對點模式的消費者沒有區別,換換監聽物件destination的值就行。接下來測試釋出訂閱模式。

測試釋出訂閱模式

建立Junit測試用例:

@Test
  public void topicTest() {
    // 設定話題監聽者,可以自由切換
    Destination destination = new ActiveMQTopic("topicListener2");
    for (int i = 0; i < 6; i++) {
      publisher.publish(destination,"Topic Message " + i);
    }
    try {
      Thread.sleep(300);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    System.out.println("使執行緒睡 300 毫秒,保證消費者消費完畢!");
  }

此處設定的訂閱者是topicListener2,讀者可以切換為topicListener1。釋出/訂閱模式和點對點模式的生產者的程式碼主要區別就是Destination的建立方式,點對點模式是呼叫new ActiveMQQueue (QUEUE_NAME),而釋出/訂閱模式是呼叫new ActiveMQTopic (QUEUE_NAME)。

執行結果:

Subscriber2 佇列收到的報文:Topic Message 0
Subscriber2 佇列收到的報文:Topic Message 1
Subscriber2 佇列收到的報文:Topic Message 2
Subscriber2 佇列收到的報文:Topic Message 3
Subscriber2 佇列收到的報文:Topic Message 4
Subscriber2 佇列收到的報文:Topic Message 5

使執行緒睡 300 毫秒,保證消費者消費完畢!

以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支援我們。