ActiveMQ学习2：SpringBoot整合ActiveMQ

红太狼 2020-11-15 13:49 761阅读 0赞

## 1.简介： ##

> 这篇博客主要记录SpringBoot整合ActiveMQ的过程，并且实现queue、Topic模式的消息传递。  
> 项目主要类如下，Email是一个消息实体，DestinationList是消息目的地列表，JmsConfiguration是一些配置，Consumer与Producer是消费者与生产者。

![在这里插入图片描述][20190927103833370.png]

## 2.实现过程： ##

> 这里只贴出ActiveMQ的相关代码，Springboot请自行配置。

### pom.xml ###

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-activemq</artifactId>
                <version>2.1.8.RELEASE</version>
            </dependency>

### application.properties ###

#========activemq
    spring.activemq.broker-url=tcp://127.0.0.1:61616
    spring.activemq.user=admin
    spring.activemq.password=admin
    #从ActiveMQ 5.12.2 开始，为了增强这个框架的安全性，ActiveMQ将强制用户配置可序列化的包名
    spring.activemq.packages.trust-all=true

### Email ###

import java.io.Serializable;
    
    /**
     * 邮件实体
     */
    public class Email implements Serializable {
    
        private static final long serialVersionUID = 7534650859786461723L;
    
    
        String from;
        String to;
        String topic;
        String content;
    
        public String getFrom() {
            return from;
        }
    
        public void setFrom(String from) {
            this.from = from;
        }
    
        public String getTo() {
            return to;
        }
    
        public void setTo(String to) {
            this.to = to;
        }
    
        public String getTopic() {
            return topic;
        }
    
        public void setTopic(String topic) {
            this.topic = topic;
        }
    
        public String getContent() {
            return content;
        }
    
        public void setContent(String content) {
            this.content = content;
        }
    
        @Override
        public String toString() {
            return "Email{" +
                    "from='" + from + '\'' +
                    ", to='" + to + '\'' +
                    ", topic='" + topic + '\'' +
                    ", content='" + content + '\'' +
                    '}';
        }
    }

### DestinationList ###

import org.apache.activemq.command.ActiveMQQueue;
    import org.apache.activemq.command.ActiveMQTopic;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration; 
    
    /**
     * 消息目的地列表
     */
    @Configuration
    public class DestinationList {
    
        final static String test_queue="test_queue";
    
        final static String test_topic="test_topic";
    
    
        @Bean(name=test_queue)
        public ActiveMQQueue test_queue(){
            return new ActiveMQQueue(test_queue);
        }
    
        @Bean(name=test_topic)
        public ActiveMQTopic test_topic(){
            return new ActiveMQTopic(test_topic);
        }
    }

### JmsConfiguration ###

import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.jms.annotation.EnableJms;
    import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
    import org.springframework.jms.config.JmsListenerContainerFactory;
    import org.springframework.jms.support.converter.MappingJackson2MessageConverter;
    import org.springframework.jms.support.converter.MessageConverter;
    import org.springframework.jms.support.converter.MessageType;
    
    import javax.jms.ConnectionFactory;
    
    /**
     * 配置Queue与Topoic并存
     */
    @Configuration
    public class JmsConfiguration {
    
        // topic模式的ListenerContainer
        @Bean(name = "jmsListenerContainerTopic")
        public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ConnectionFactory connectionFactory) {
            DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
            bean.setPubSubDomain(true);
            bean.setConnectionFactory(connectionFactory);
            return bean;
        }
    
        // queue模式的ListenerContainer
        @Bean(name = "jmsListenerContainerQueue")
        public JmsListenerContainerFactory<?> jmsListenerContainerQueue(ConnectionFactory connectionFactory) {
            DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
            bean.setConnectionFactory(connectionFactory);
            return bean;
        }
    
    }

### Producer ###

import org.apache.activemq.command.ActiveMQQueue;
    import org.apache.activemq.command.ActiveMQTopic;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsMessagingTemplate;
    import org.springframework.messaging.core.MessagePostProcessor;
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import javax.annotation.Resource;
    import javax.jms.Destination;
    import javax.jms.Topic;
    import java.io.Serializable;
    import java.util.Queue;
    
    /*
     *消息生产者
     */
    @RestController
    public class Producer {
        @Resource
        private JmsMessagingTemplate jmsMessagingTemplate;
    
        @Autowired
        ActiveMQQueue test_queue;
    
        @Autowired
        ActiveMQTopic test_topic;
    
        //生产消息
        @RequestMapping("sendQueueMsg")
        public String sendQueueMsg(){
            Email email = new Email();
            email.setFrom("A");
            email.setFrom("B");
            email.setTopic("hello");
            email.setContent("How are you??");
            jmsMessagingTemplate.convertAndSend( test_queue,email);
            return "Queue生产者发生消息："+email;
        }
    
    
        //生产消息
        @RequestMapping("sendTopicMsg")
        public String sendTopicMsg(){
            Email email = new Email();
            email.setFrom("A");
            email.setFrom("B");
            email.setTopic("hello");
            email.setContent("How are you??");
            jmsMessagingTemplate.convertAndSend( test_topic,email);
            return "Topic生产者发生消息："+email;
        }
    
    }

### Consumer ###

/**
     * 消费者
     *
     */
    @Component
    public class Consumer {
    
        /**
         * destination:是消息目的地
         * containerFactory：容器类型，需要指定JmsConfiguration类中的Bean
         * @param msg
         */
        @JmsListener(destination=DestinationList.test_queue,containerFactory = "jmsListenerContainerQueue")
        public void msgConsumer(Serializable msg){
            Email email =  (Email)msg;
            System.out.println("Queue消费者接收到消息："+ email.toString());
        }
    
    
        @JmsListener(destination=DestinationList.test_topic,containerFactory = "jmsListenerContainerTopic")
        public void msgConsumerTopic1(Serializable msg){
            Email email =  (Email)msg;
            System.out.println("Topic消费者1接收到消息："+ email.toString());
        }
    
        @JmsListener(destination=DestinationList.test_topic,containerFactory = "jmsListenerContainerTopic")
        public void msgConsumerTopic2(Serializable msg){
            Email email =  (Email)msg;
            System.out.println("Topic消费者2接收到消息："+ email.toString());
        }
    
    }

[20190927103833370.png]: /images/1605448127334.png