2. 软件开发
    3. springboot 2.1 实践教程（十八)-整合ActiveMQ

    springboot 2.1 实践教程（十八)-整合ActiveMQ

    xiaoxiao2025-08-11  6

    ActiveMQ介绍

    ActiveMQ是JMS规范的具体实现；它是Apache下的一个项目，采用Java语言开发；是一款流行的开源消息服务器.支持多种语言编写客户端，支持多种传输协议：TCP,SSL,NIO,UDP等

     

    安装ActiveMQ

    具体如何安装本章节就不做介绍了，请自行百度安装，比较简单。

    下面是Spring Boot 整合ActiveMQ的完整代码

    pom.xml增加ActiveMQ的依赖

     

    <!-- activemq--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-activemq</artifactId> </dependency> <!--消息队列连接池--> <dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-pool</artifactId> <version>5.15.0</version> </dependency> 配置Application.yml spring: #activemq activemq: broker-url: tcp://127.0.0.1:61616 in-memory: false pool: #true表示使用连接池 enabled: true #连接池最大连接数 max-connections: 10 ActiveMQConfig package org.learn.config; import org.apache.activemq.ActiveMQConnectionFactory; 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.context.annotation.Configuration; import org.springframework.jms.config.DefaultJmsListenerContainerFactory; import org.springframework.jms.config.JmsListenerContainerFactory; import javax.jms.Queue; import javax.jms.Topic; @Configuration public class ActiveMQConfig { @Value("${spring.activemq.broker-url}") private String brokerUrl; @Bean public Queue queue() { return new ActiveMQQueue("ActiveMQQueue"); } @Bean public Topic topic(){ return new ActiveMQTopic("ActiveMQTopic"); } @Bean public ActiveMQConnectionFactory connectionFactory() { return new ActiveMQConnectionFactory(brokerUrl); } //Queue模式连接注入 @Bean public JmsListenerContainerFactory<?> jmsListenerContainerQueue(ActiveMQConnectionFactory connectionFactory){ DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory(); bean.setConnectionFactory(connectionFactory); return bean; } //Topic模式连接注入 @Bean public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ActiveMQConnectionFactory connectionFactory){ DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory(); //设置为发布订阅方式, 默认情况下使用的生产消费者方式 bean.setPubSubDomain(true); bean.setConnectionFactory(connectionFactory); return bean; } } ProducerController

     

    package org.learn.controller; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jms.core.JmsMessagingTemplate; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.ResponseBody; import javax.jms.Queue; import javax.jms.Topic; @Controller public class ProducerController { //注入点对点的模式（Queue模式） @Autowired private Queue queue; //注入订阅模式（Topic）的消息 @Autowired private Topic topic; //注入springboot封装的工具类 @Autowired private JmsMessagingTemplate jms; /** * 点对点模式（queue）模式发消息 * @param text */ @GetMapping("/queueSend") @ResponseBody public String queueSend(String text) { //发送消息至消息中间件代理（Broker） jms.convertAndSend(queue, text); return "success"; } /** * 订阅模式（topic）发送消息 * @param text * @return */ @RequestMapping("/topicSend") public String topicSend(String text){ jms.convertAndSend(topic,text); return "topic 发送成功"; } }

    以上就完成了生产者的代码，包括了两种消息模式，分别是点对点模式和订阅发布模式。

     

    下面完成消费者的代码

    首先是点对点模式（Queue）的代码

     

    package org.learn.service; import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; /** * 消息消費者(Queue模式) */ @Component public class ConsumerQueue { // 使用JmsListener配置消费者监听的队列，其中text是接收到的消息 @JmsListener(destination = "ActiveMQQueue",containerFactory = "jmsListenerContainerQueue") public void receiveQueue(String text) { System.out.println("消息消費者收到的报文为:"+text); } }

     

    订阅发布模式（Topic）的代码 package org.learn.service; import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; /** * 消息消費者（TopicM模式） */ @Component public class ConsumerTopic { // 使用JmsListener配置消费者监听的队列，其中text是接收到的消息 @JmsListener(destination = "ActiveMQTopic", containerFactory = "jmsListenerContainerTopic") public void receiveQueue(String text) { System.out.println("消息消費者收到的Topic报文为:" + text); } }

     

    注意@JmsListener这个注解，包括两个参数destination是我们在Config类里创建的destination（目的地），如果是Queue模式就配置ActiveMQQueue，如果是Topic模式就配置ActiveMQTopic，第二个参数是containerFactory，用于区别两种不同模式的连接

    以上就完成了Spring Boot 整合ActiveMQ的全部代码了，下面我们进行测试

    首先测试点对点模式（Queue）

    发送请求

    http://localhost:8080/queueSend?text=hello,activemq---Queue

    控制台输出结果

     

     

    测试发布订阅模式（Topic）

    发送请求

    http://localhost:8080/topicSend?text=hello,activemq---Topic

    控制台输出结果

     

    测试成功，生产者发送消息，消费者监听消息，如果有消息发送，消费者就会去获取到消息。

     

    问题延伸：现在我们成功完成了消息的发送与接收，但是大家有没有想到，目前的程序 ，虽然消息已经被消费者接收，但是我们的生产者如何知道这个消息已经成功接收呢？

    解决办法：实现双向绑定，在消费者接收到信息后给生产者返回一个内容，告诉生产者已经接收到消息，废话不多说，开始改造代码

    改造ConsumerQueue

     

    package org.learn.service; import org.springframework.jms.annotation.JmsListener; import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.stereotype.Component; /** * 消息消費者(Queue模式) */ @Component public class ConsumerQueue { // 使用JmsListener配置消费者监听的队列，其中text是接收到的消息 @JmsListener(destination = "ActiveMQQueue",containerFactory = "jmsListenerContainerQueue") @SendTo("out.queue") public String receiveQueue(String text) { System.out.println("消息消費者收到的报文为:"+text); return" queue---消息已经收到，over!"; } }

    增加了注释@SendTo("out.queue")，表示消费者收到消息后讲返回相应的字符串至out.queue这个队列中。

    改造ConsumerTopic package org.learn.service; import org.springframework.jms.annotation.JmsListener; import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.stereotype.Component; /** * 消息消費者（TopicM模式） */ @Component public class ConsumerTopic { // 使用JmsListener配置消费者监听的队列，其中text是接收到的消息 @JmsListener(destination = "ActiveMQTopic", containerFactory = "jmsListenerContainerTopic") @SendTo("out.topic") public String receiveQueue(String text) { System.out.println("消息消費者收到的Topic报文为:" + text); return" topic---消息已经收到，over!"; } }

     

    增加ProducerListener package org.learn.service; import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; /** * 监听消费者的反馈（Queue模式） */ @Component public class ProducerListener { @JmsListener(destination = "out.queue") public void consumerMessage(String text) { System.out.println("从out.queue队列收到的回复报文为:" + text); } @JmsListener(destination = "out.topic", containerFactory = "jmsListenerContainerTopic") public void consumerTopicMessage(String text) { System.out.println("从out.topic队列收到的回复报文为:" + text); } } 增加ProducerTopicListener package org.learn.service; import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; /** * 监听消费者的反馈(Topic模式) */ @Component public class ProducerTopicListener { @JmsListener(destination = "out.topic") public void consumerMessage(String text) { System.out.println("从out.topic队列收到的回复报文为:" + text); } }

     

    控制台输出的结果

     

    最新回复(0)