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

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

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

和點(diǎn)對(duì)點(diǎn)方式不同，發(fā)布到topic的消息會(huì)被所有訂閱者消費(fèi)；當(dāng)生產(chǎn)者發(fā)布消息時(shí)，不管是否有消費(fèi)者，都不會(huì)保存消息；一定要先有消息的消費(fèi)者，后有消息的生產(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:61616spring.activemq.in-memory=truespring.activemq.pool.enabled=falsespring.activemq.password=adminspring.activemq.user=admin#默認(rèn)值false，表示point to point（點(diǎn)到點(diǎn)）模式，true時(shí)代表發(fā)布訂閱模式，需要手動(dòng)開啟#spring.jms.pub-sub-domain=true

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

import org.springframework.beans.factory.annotation.Autowired;import org.springframework.jms.core.JmsMessagingTemplate;import org.springframework.stereotype.Service;import javax.jms.Destination;/** * 生產(chǎn)者 */@Servicepublic 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;/** * 消費(fèi)者 */@Servicepublic class Subscriber2 { private static Logger logger = LoggerFactory.getLogger(Subscriber2.class); @JmsListener(destination = 'topicListener2') public void subscriber(String text) { logger.info('Subscriber2 收到的報(bào)文：{}', text); }}

import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.jms.annotation.JmsListener;import org.springframework.stereotype.Component;import javax.jms.JMSException;/** * 消費(fèi)者 */@Componentpublic 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 收到的報(bào)文：{}', text); }}

發(fā)布訂閱模式和點(diǎn)對(duì)點(diǎn)模式的消費(fèi)者沒有區(qū)別，換換監(jiān)聽對(duì)象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 毫秒，保證消費(fèi)者消費(fèi)完畢!'); }

此處設(shè)置的訂閱者是topicListener2，讀者可以切換為topicListener1。發(fā)布/訂閱模式和點(diǎn)對(duì)點(diǎn)模式的生產(chǎn)者的代碼主要區(qū)別就是Destination的創(chuàng)建方式，點(diǎn)對(duì)點(diǎn)模式是調(diào)用new ActiveMQQueue (QUEUE_NAME)，而發(fā)布/訂閱模式是調(diào)用new ActiveMQTopic (QUEUE_NAME)。

執(zhí)行結(jié)果：

Subscriber2 隊(duì)列收到的報(bào)文：Topic Message 0Subscriber2 隊(duì)列收到的報(bào)文：Topic Message 1Subscriber2 隊(duì)列收到的報(bào)文：Topic Message 2Subscriber2 隊(duì)列收到的報(bào)文：Topic Message 3Subscriber2 隊(duì)列收到的報(bào)文：Topic Message 4Subscriber2 隊(duì)列收到的報(bào)文：Topic Message 5

使線程睡 300 毫秒，保證消費(fèi)者消費(fèi)完畢!

以上就是本文的全部內(nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持好吧啦網(wǎng)。