Spring Boot ActiveMQ发布/订阅消息模式原理解析

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.sthttp://ereotype.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 毫秒，保证消费者消费完毕!

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。