Spring Boot ActiveMQ發(fā)布/訂閱消息模式原理解析

更新時間：2020年07月02日 12:04:18 作者：樓蘭的胡楊

這篇文章主要介紹了Spring Boot ActiveMQ發(fā)布/訂閱消息模式原理解析,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下

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

發(fā)布/訂閱模式的工作示意圖

消息生產(chǎn)者將消息（發(fā)布）到topic中，可以同時有多個消息消費者（訂閱）消費該消息。

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

軟件環(huán)境

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時代表發(fā)布訂閱模式，需要手動開啟
#spring.jms.pub-sub-domain=true

創(chuàng)建生產(chǎn)者和消費者

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

import javax.jms.Destination;

/**
 * 生產(chǎn)者
 */
@Service
public class Publisher {
  @Autowired
  private JmsMessagingTemplate jmsMsgTemplate;

  /**
   * 發(fā)送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);
  }

}

發(fā)布訂閱模式和點對點模式的消費者沒有區(qū)別，換換監(jiān)聽對象destination的值就行。接下來測試發(fā)布訂閱模式。

測試發(fā)布訂閱模式

創(chuàng)建Junit測試用例：

@Test
  public void topicTest() {
    // 設置話題監(jiān)聽者，可以自由切換
    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。發(fā)布/訂閱模式和點對點模式的生產(chǎn)者的代碼主要區(qū)別就是Destination的創(chuàng)建方式，點對點模式是調用new ActiveMQQueue (QUEUE_NAME)，而發(fā)布/訂閱模式是調用new ActiveMQTopic (QUEUE_NAME)。

執(zhí)行結果：

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 毫秒，保證消費者消費完畢!

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: