Spring Boot ActiveMQ發佈/訂閱消息模式原理解析

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

發佈/訂閱模式的工作示意圖

消息生產者將消息（發佈）到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 毫秒，保證消費者消費完畢!