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

發(fā)布/訂閱模式的工作示意圖
消息生產(chǎn)者將消息(發(fā)布)到topic中,可以同時有多個消息消費者(訂閱)消費該消息。
和點對點方式不同,發(fā)布到topic的消息會被所有訂閱者消費;當(dāng)生產(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() {
// 設(shè)置話題監(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 毫秒,保證消費者消費完畢!");
}
此處設(shè)置的訂閱者是topicListener2,讀者可以切換為topicListener1。發(fā)布/訂閱模式和點對點模式的生產(chǎn)者的代碼主要區(qū)別就是Destination的創(chuàng)建方式,點對點模式是調(diào)用new ActiveMQQueue (QUEUE_NAME),而發(fā)布/訂閱模式是調(diào)用new ActiveMQTopic (QUEUE_NAME)。
執(zhí)行結(jié)果:
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 毫秒,保證消費者消費完畢!
以上就是本文的全部內(nèi)容,希望對大家的學(xué)習(xí)有所幫助,也希望大家多多支持腳本之家。
相關(guān)文章
Java使用cxf發(fā)布及調(diào)用webservice接口的方法詳解
今天小編就為大家分享一篇關(guān)于Java使用CXF發(fā)布及調(diào)用WebService接口的方法,文中通過代碼示例給大家介紹的非常詳細,具有很好的參考價值,需要的朋友一起跟隨小編來看看吧2023-08-08
Java連接合并2個數(shù)組(Array)的5種方法例子
最近在寫代碼時遇到了需要合并兩個數(shù)組的需求,突然發(fā)現(xiàn)以前沒用過,于是研究了一下合并數(shù)組的方式,這篇文章主要給大家介紹了關(guān)于Java連接合并2個數(shù)組(Array)的5種方法,需要的朋友可以參考下2023-12-12
SpringBoot3.x中自定義開發(fā)通用SDK的實現(xiàn)
本文介紹了SpringBoot和Maven創(chuàng)建自定義的SDK,包括創(chuàng)建項目、修改配置、編寫配置類、設(shè)置配置文件、構(gòu)建Jar包等,具有一定的參考價值,感興趣的可以了解一下2024-12-12
SpringBoot是如何實現(xiàn)自動配置的你知道嗎
這篇文章主要介紹了詳解SpringBoot自動配置原理,小編覺得挺不錯的,現(xiàn)在分享給大家,也給大家做個參考。一起跟隨小編過來看看吧2021-08-08
springboot使用Gateway做網(wǎng)關(guān)并且配置全局攔截器的方式
本文介紹了Spring Cloud Gateway的使用,包括為什么需要使用網(wǎng)關(guān)、如何簡單使用、斷言工廠路由、路由過濾器、全局過濾器以及跨域請求配置,感興趣的朋友一起看看吧2025-02-02

