【Spring Boot】ActiveMQ 發布/訂閱消息模式介紹

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

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

   消息生產者將消息（發布）到topic中，可以同時有多個消息消費者（訂閱）消費該消息。

   和點對點方式不同，發布到topic的消息會被所有訂閱者消費；當生產者發布消息時，不管是否有消費者，都不會保存消息；一定要先有消息的消費者，后有消息的生產者。

軟件環境

1. ActiveMQ 5.15.13
2. java version 13.0.1
3. IntelliJ IDEA 2019.3.2 (Ultimate Edition)
4. 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 毫秒，保證消費者消費完畢!

   以上就是這篇文章的全部內容了，希望本文對大家的學習或者工作能帶來一定的幫助，如有疑問請留言交流。祝各位生活愉快！