详解Springboot整合ActiveMQQueue和Topic两种模式

这篇文章主要介绍了详解Springboot整合ActiveMQ（Queue和Topic两种模式），文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧

ActiveMQ简介

1、ActiveMQ简介

Apache ActiveMQ是Apache软件基金会所研发的开放源代码消息中间件；由于ActiveMQ是一个纯Java程序，因此只需要操作系统支持Java虚拟机，ActiveMQ便可执行。

2、ActiveMQ下载

下载地址：http://activemq.apache.org/components/classic/download/

下载完成后解压双击activemq.bat文件打开(不用安装，直接使用)，目录和打开后效果如下：

运行后，浏览器访问http://localhost:8161/地址进入一下界面。

点击Manage ActiveMQ broker登录到ActiveMQ管理页面，默认账号和密码都是admin。管理页面如下：

SpringBoot整合ActiveMQ

1、新建SpringBoot项目

新建Springboot项目，添加对应的依赖。项目完整的pom.xml文件如下：

4.0.0org.springframework.bootspring-boot-starter-parent2.2.5.RELEASEcom.mcyspringboot-mq0.0.1-SNAPSHOTspringboot-mqDemo project for Spring Boot1.8org.springframework.bootspring-boot-starter-weborg.springframework.bootspring-boot-starter-activemqorg.springframework.bootspring-boot-starter-testtestorg.junit.vintagejunit-vintage-engineorg.springframework.bootspring-boot-maven-plugin

2、项目结构

3、相关配置信息

在application.properties类中添加ActiveMQ相关的配置信息

server.port=8080 server.servlet.context-path=/mq #MQ服务器地址 spring.activemq.broker-url=tcp://localhost:61616 #用户名 spring.activemq.user=admin #密码 spring.activemq.password=admin #设置是queue队列还是Topic，false为Queue，true为Topic，默认false-Queue spring.jms.pub-sub-domain=false #spring.jms.pub-sub-domain=true #变量，定义队列和topic的名称 myqueue: activemq-queue mytopic: activemq-topic

4、ActiveMQ配置类

ActiveMQ配置类ConfigBean，配置了queue队列和topic两种模式，代码如下：

import org.apache.activemq.command.ActiveMQQueue; import org.apache.activemq.command.ActiveMQTopic; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.jms.annotation.EnableJms; import org.springframework.stereotype.Component; import javax.jms.Topic; /** 1. MQ配置类 */ @Component @EnableJms public class ConfigBean { @Value("${myqueue}") private String myQueue; @Value("${mytopic}") private String topicName; //队列 @Bean public ActiveMQQueue queue(){ return new ActiveMQQueue(myQueue); } //topic @Bean public Topic topic(){ return new ActiveMQTopic(topicName); } }

queue队列模式

队列模式即点对点传输。

点对点消息传递域的特点如下：

每个消息只能有一个消费者，类似于1对1的关系。好比个人快递自己领自己的。

消息的生产者和消费者之间没有时间上的相关性。无论消费者在生产者发送消息的时候是否处于运行状态，消费者都可以提取消息。好比我们的发送短信，发送者发送后不见得接收者会即收即看。

消息被消费后队列中不会再存储，所以消费者不会消费到已经被消费掉的消息。

1、队列生产者

QueueProducerController类为队列生产者控制器，主要向消息队列中发送消息。代码如下：

import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jms.core.JmsMessagingTemplate; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import javax.jms.Queue; /* * 队列消息生产者 */ @RestController public class QueueProducerController { @Autowired private JmsMessagingTemplate jmsMessagingTemplate; @Autowired private Queue queue; /* * 消息生产者 */ @RequestMapping("/sendmsg") public void sendmsg(String msg) { System.out.println("发送消息到队列：" + msg); // 指定消息发送的目的地及内容 this.jmsMessagingTemplate.convertAndSend(this.queue, msg); } }

2、队列消费者

QueueConsumerController类为队列消费者控制器，具体代码如下：

import org.springframework.beans.factory.annotation.Value; import org.springframework.jms.annotation.JmsListener; import org.springframework.web.bind.annotation.RestController; /* 1. 队列queue消费者控制器 */ @RestController public class QueueConsumerController { /* * 消费者接收消息 */ @JmsListener(destination="${myqueue}") public void readActiveQueue(String message) { System.out.println("接受到：" + message); } }

3、测试效果

运行项目在浏览器中访问http://localhost:8080/mq/sendmsg?msg=123。向消息队列中发送123。控制台输出效果：

ActiveMQ控制台显示：

Number Of Pending Messages：消息队列中待处理的消息

Number Of Consumers：消费者的数量

Messages Enqueued：累计进入过消息队列的总量

Messages Dequeued：累计消费过的消息总量

【注】队列模式时，配置文件application.properties中spring.jms.pub-sub-domain属性必须设置为false。

Topic模式

topic模式基于发布/订阅模式的传输。

基于发布/订阅模式的传输的特点如下：

生产者将消息发布到topic中，每个消息可以有多个消费者，属于1：N的关系；

生产者和消费者之间有时间上的相关性。订阅某一个主题的消费者只能消费自它订阅之后发布的消息。

生产者生产时，topic不保存消息它是无状态的不落地，假如无人订阅就去生产，那就是一条废消息。

1、topic生产者

TopicProducerController类为topic生产者控制器，主要向消息队列中发送消息。代码如下：

import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jms.core.JmsMessagingTemplate; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import javax.jms.Queue; import javax.jms.Topic; /* * topic消息生产者 */ @RestController public class TopicProducerController { @Autowired private JmsMessagingTemplate jmsMessagingTemplate; @Autowired private Topic topic; /* * 消息生产者 */ @RequestMapping("/topicsendmsg") public void sendmsg(String msg) { System.out.println("发送消息到MQ：" + msg); // 指定消息发送的目的地及内容 this.jmsMessagingTemplate.convertAndSend(this.topic, msg); } }

2、topic消费者

TopicConsumerController类为topic消费者控制器，其中写了两个消费者方法，可以理解为有两个用户订阅。具体代码如下：

import org.springframework.jms.annotation.JmsListener; import org.springframework.web.bind.annotation.RestController; /* 1. topic消费者控制器 */ @RestController public class TopicConsumerController { /* * 消费者接收消息 */ @JmsListener(destination="${mytopic}") public void readActiveQueue(String message) { System.out.println("接受到：" + message); } @JmsListener(destination="${mytopic}") public void readActiveQueue1(String message) { System.out.println("接受到：" + message); } }