ActiveMQ实战（SpringBoot）

蔚落 2021-12-12 13:34 201阅读 0赞

> 本篇文章是实战篇，有关MQ的基本概念请自行查阅有关资料，网上关于此方面教程也多如牛毛，本章就不赘述了。
> 
> 有关ActiveMQ的介绍可以看这篇教程：[ActiveMQ入门][ActiveMQ]
> 
>  示例项目需要使用SpringBoot来搭建两个项目，jms项目作为消息发送端，jms-client项目作为消息接收端。

## 一、安装ActiveMQ ##

[ActiveMq下载安装教程][ActiveMq]，读者可参照此教程下载安装ActiveMQ。

## 二、JMS项目--消息发送端 ##

###  1、POM文件 ###

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.1.1.RELEASE</version>
            <relativePath/> 
        </parent>
        <groupId>com.scb</groupId>
        <artifactId>jms</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>jms</name>
        <description>Demo project for Spring Boot</description>
    
        <properties>
            <java.version>1.8</java.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-activemq</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <optional>true</optional>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    
    </project>

其中，spring-boot-starter-activemq就是ActiveMQ需要加载的依赖。

### 2、配置ActiveMQ ###

package com.scb.jms.config;
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.activemq.command.ActiveMQQueue;
    import org.apache.activemq.command.ActiveMQTopic;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.jms.connection.CachingConnectionFactory;
    import org.springframework.jms.core.JmsTemplate;
    
    import javax.jms.ConnectionFactory;
    
    @Slf4j
    @Configuration
    public class JmsConfig {
        //Create ConnectionFactory Bean
        @Bean
        public ActiveMQConnectionFactory connectionFactory(){
            ActiveMQConnectionFactory connectionFactory=new ActiveMQConnectionFactory();
            connectionFactory.setBrokerURL("tcp://localhost:61616");
            return connectionFactory;
        }
    
        // Create cache
        @Bean
        public CachingConnectionFactory cachingConnectionFactory(ConnectionFactory connectionFactory){
            CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
            cachingConnectionFactory.setTargetConnectionFactory(connectionFactory);
            cachingConnectionFactory.setSessionCacheSize(10);
            return cachingConnectionFactory;
        }
    
        //Create Destination
        @Bean
        public ActiveMQQueue queue(){
            ActiveMQQueue queue=new ActiveMQQueue();
            queue.setPhysicalName("jms.queue");
            return queue;
        }
    
        @Bean
        public ActiveMQTopic topic(){
            ActiveMQTopic topic = new ActiveMQTopic();
            topic.setPhysicalName("jms.topic");
            return topic;
        }
    
        //Create JmsTemplate Bean queue
        @Bean
        public JmsTemplate queueJmsTemplate(ConnectionFactory cachingConnectionFactory,ActiveMQQueue queue){
            log.info("queueJmsTemplate isQueue :: {}",queue.isQueue());
            JmsTemplate jmsTemplate=new JmsTemplate();
            jmsTemplate.setConnectionFactory(cachingConnectionFactory);
            jmsTemplate.setDefaultDestination(queue);
            return jmsTemplate;
        }
    
        //Create JmsTemplate Bean topic
        @Bean
        public JmsTemplate topicJmsTemplate(ConnectionFactory cachingConnectionFactory,ActiveMQTopic topic){
            log.info("topicJmsTemplate isTopic :: {}",topic.isTopic());
            JmsTemplate jmsTemplate=new JmsTemplate();
            jmsTemplate.setConnectionFactory(cachingConnectionFactory);
            jmsTemplate.setDefaultDestination(topic);
            jmsTemplate.setPubSubDomain(true);
            return jmsTemplate;
        }
    }

首先，需要先配置ActiveMQ的连接工厂（ActiveMQConnectionFactory），通过该连接工厂我们就能创建ActiveMQConnection。但是频繁的创建销毁是非常浪费系统资源的，所以这里我们将ActiveMQConnectionFactory配置到CachingConnectionFactory中，这相当于是一个连接池。

然后，我们创建了两种Destination：Queue和Topic。最后创建JmsTemplate对象。通过JmsTemplate，我们可以非常简单地去发送接收消息。

### 3、Service层 ###

在配置好ActiveMQ后，我们就可以通过JmsTemplate来进行消息的发送了。这里，为了解耦，将业务逻辑封装进Service层。

package com.scb.jms.service;
    
    import javax.jms.JMSException;
    
    public interface ISenderService {
        void sendMessage(String message);
        void sendMessages() throws JMSException;
    }

上面是发送消息的服务接口，下面我们看看，它的两个实现类。

package com.scb.jms.service;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.stereotype.Service;
    
    import javax.jms.JMSException;
    import java.util.Date;
    
    @Slf4j
    @Service
    public class QueueSenderServiceImpl implements ISenderService {
        @Autowired
        @Qualifier(value = "queueJmsTemplate")
        private JmsTemplate jmsTemplate;
    
        @Override
        public void sendMessage(String message) {
            jmsTemplate.convertAndSend(message);
        }
    
        public void sendMessages() throws JMSException {
            StringBuilder payload = null;
            for (int i = 0; i < 100; ++i) {
                payload = new StringBuilder();
                payload.append("Message [").append(i).append("] sent at: ").append(new Date());
                jmsTemplate.convertAndSend(payload.toString());
                log.info("Sending message number [" + i + "]");
            }
        }
    }

QueueSenderServiceImpl类使用queueJmsTemplate来将消息发送到Queue中。

package com.scb.jms.service;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.stereotype.Service;
    
    import javax.jms.JMSException;
    import java.util.Date;
    
    @Slf4j
    @Service
    public class TopicSenderServiceImpl implements ISenderService {
        @Autowired
        @Qualifier(value = "topicJmsTemplate")
        private JmsTemplate jmsTemplate;
    
        @Override
        public void sendMessage(String message) {
            jmsTemplate.convertAndSend(message);
        }
    
        public void sendMessages() throws JMSException {
            StringBuilder payload = null;
            for (int i = 0; i < 100; ++i) {
                payload = new StringBuilder();
                payload.append("Message [").append(i).append("] sent at: ").append(new Date());
                jmsTemplate.convertAndSend(payload.toString());
                log.info("Sending message number [" + i + "]");
            }
        }
    }

而TopicSenderServiceImpl类使用topicJmsTemplate来将消息发送到Topic中。

通过jmsTemplate的convertAndSend方法，我们可以非常简单地进行消息发送。

### 4、Controller层 ###

Service层开发完成后，我们就可以在Controller层进行调用了。

package com.scb.jms.controller;
    
    import com.scb.jms.service.ISenderService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.ResponseBody;
    
    import javax.jms.JMSException;
    
    @Controller
    @RequestMapping(value = "/jms/topic")
    public class JmsTopicController {
        @Autowired
        @Qualifier(value = "topicSenderServiceImpl")
        private ISenderService senderService;
    
        @GetMapping("/{msg}")
        public void sendMessage(@PathVariable(name = "msg") String msg) {
            senderService.sendMessage(msg);
        }
    
        @GetMapping("/")
        @ResponseBody
        public void sendMessages() {
            try {
                senderService.sendMessages();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

这里，也拆分成两个Controller。

package com.scb.jms.controller;
    
    import com.scb.jms.service.ISenderService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.ResponseBody;
    
    import javax.jms.JMSException;
    
    @Controller
    @RequestMapping(value = "/jms/queue")
    public class JmsQueueController {
        @Autowired
        @Qualifier(value = "queueSenderServiceImpl")
        private ISenderService senderService;
    
        @GetMapping("/{msg}")
        public void sendMessage(@PathVariable(name = "msg") String msg){
            senderService.sendMessage(msg);
        }
    
        @GetMapping("/")
        @ResponseBody
        public void sendMessages(){
            try {
                senderService.sendMessages();
            }catch (JMSException e){
                e.printStackTrace();
            }
        }
    }

OK,到这里我们jms项目消息发送端已经开发完成。如果你已经把ActiveMQ运行起来了，当你访问[http://localhost:8080/jms/topic/][http_localhost_8080_jms_topic]就可以将消息发送到ActiveMQ的Topic中，如下图所示。（其他同理）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NzcxNDc1_size_16_color_FFFFFF_t_70][]

### 三、jms-client项目--消息接收端 ###

### 1、pom文件 ###

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.1.1.RELEASE</version>
            <relativePath/> 
        </parent>
        <groupId>com.scb</groupId>
        <artifactId>jms_client</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>jms_client</name>
        <description>Demo project for Spring Boot</description>
    
        <properties>
            <java.version>1.8</java.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-activemq</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <optional>true</optional>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
    
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    
    </project>

可以看到依赖包和jms项目是一样的

### 2、ActiveMQ配置类 ###

package com.scb.jms_client.config;
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.activemq.command.ActiveMQQueue;
    import org.apache.activemq.command.ActiveMQTopic;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.jms.connection.CachingConnectionFactory;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.listener.DefaultMessageListenerContainer;
    
    import javax.jms.ConnectionFactory;
    import javax.jms.MessageListener;
    
    @Slf4j
    @Configuration
    public class JmsConfig {
        //Create ConnectionFactory Bean
        @Bean
        public ActiveMQConnectionFactory connectionFactory(){
            ActiveMQConnectionFactory connectionFactory=new ActiveMQConnectionFactory();
            connectionFactory.setBrokerURL("tcp://localhost:61616");
            return connectionFactory;
        }
    
        // Create cache
        @Bean
        public CachingConnectionFactory cachingConnectionFactory(ConnectionFactory connectionFactory){
            CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
            cachingConnectionFactory.setTargetConnectionFactory(connectionFactory);
            cachingConnectionFactory.setSessionCacheSize(10);
            return cachingConnectionFactory;
        }
    
        //Create Destination
        @Bean
        public ActiveMQQueue queue(){
            ActiveMQQueue queue=new ActiveMQQueue();
            queue.setPhysicalName("jms.queue");
            return queue;
        }
    
        @Bean
        public ActiveMQTopic topic(){
            ActiveMQTopic topic = new ActiveMQTopic();
            topic.setPhysicalName("jms.topic");
            return topic;
        }
    
        //Create JmsTemplate Bean queue
        @Bean
        public JmsTemplate queueJmsTemplate(ConnectionFactory cachingConnectionFactory,ActiveMQQueue queue){
            log.info("jmsTemplate:"+cachingConnectionFactory.getClass());
            JmsTemplate jmsTemplate=new JmsTemplate();
            jmsTemplate.setConnectionFactory(cachingConnectionFactory);
            //jmsTemplate.setDefaultDestinationName("jms.queue");
            jmsTemplate.setDefaultDestination(queue);
            return jmsTemplate;
        }
    
        //Create JmsTemplate Bean topic
        @Bean
        public JmsTemplate topicJmsTemplate(ConnectionFactory cachingConnectionFactory,ActiveMQTopic topic){
            JmsTemplate jmsTemplate=new JmsTemplate();
            jmsTemplate.setConnectionFactory(cachingConnectionFactory);
            jmsTemplate.setDefaultDestination(topic);
            jmsTemplate.setPubSubNoLocal(true);
            return jmsTemplate;
        }
    
        //配置消息侦听容器
        @Bean
        public DefaultMessageListenerContainer queueListenerContainer(ConnectionFactory cachingConnectionFactory, ActiveMQQueue queue, MessageListener messageListener){
            DefaultMessageListenerContainer listenerContainer=new DefaultMessageListenerContainer();
            listenerContainer.setConnectionFactory(cachingConnectionFactory);
            listenerContainer.setDestination(queue);
            listenerContainer.setMessageListener(messageListener);
            return listenerContainer;
        }
    
        //配置消息侦听容器
        @Bean
        public DefaultMessageListenerContainer topicListenerContainer(ConnectionFactory cachingConnectionFactory, ActiveMQTopic topic, MessageListener messageListener){
            DefaultMessageListenerContainer listenerContainer=new DefaultMessageListenerContainer();
            listenerContainer.setConnectionFactory(cachingConnectionFactory);
            listenerContainer.setDestination(topic);
            listenerContainer.setMessageListener(messageListener);
            listenerContainer.setPubSubDomain(true);
            return listenerContainer;
        }
    }

这里跟jms项目相比，多配置了DefaultMessageListenerContainer Bean。这是个消息监听容器，当配置好它后，我们就可以异步监听是否有消息。它依赖于MessageListener对象，下面我们来配置他。

package com.scb.jms_client.pojo;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.stereotype.Component;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    @Slf4j
    @Component
    public class JmsMessageListener implements MessageListener {
        @Override
        public void onMessage(Message message) {
            try {
                TextMessage msg = (TextMessage) message;
                log.info("Consumed message: " + msg.getText());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

MessageListener的onMessage定义了接收到消息时采取的措施。

需注意的是：jms-client项目端口要改成8081(避免和jms项目端口冲突)

## 四、运行 ##

运行两个项目和ActiveMQ，访问[http://localhost:8080/jms/topic/][http_localhost_8080_jms_topic]，将消息发送到topic中。jms项目的输出如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NzcxNDc1_size_16_color_FFFFFF_t_70 1][]

此时，在jms-client项目的控制台可以看到，如下输出：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NzcxNDc1_size_16_color_FFFFFF_t_70 2][]

这是因为jms-client项目配置了DefaultMessageListenerContainer，有消息进入ActiveMQ会自动进行消费。

## 后记 ##

本章介绍了如何在SpringBoot中搭建一个简单的ActiveMQ应用项目。之后，我将继续介绍ActiveMQ如何与Zookeeper进行集成，搭建ActiveMQ集群。

[ActiveMQ]: https://segmentfault.com/a/1190000014958916
[ActiveMq]: https://blog.csdn.net/qq_26462567/article/details/82977384
[http_localhost_8080_jms_topic]: http://localhost:8080/jms/topic/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NzcxNDc1_size_16_color_FFFFFF_t_70]: /images/20211211/2d3b6d435cfc43a8993e627e6ad4c105.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NzcxNDc1_size_16_color_FFFFFF_t_70 1]: /images/20211211/c8b8743c6c734a3b81501ef72aa5d1a4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NzcxNDc1_size_16_color_FFFFFF_t_70 2]: /images/20211211/04c3ab6baf4140e6b407d3428c079e67.png