尚硅谷ActiveMQ学习笔记（4）-- 整合ActiveMQ

以你之姓@ 2022-10-22 07:34 180阅读 0赞

### ActiveMQ的Broker ###

### 一、是什么？ ###

相当于一个ActiveMQ服务器实例  
说白了，Broker其实就是实现了用代码的形式启动ActiveMQ将MQ嵌入到Java代码中，以便随时用随时启动。  
在用的时候再去启动这样能节省了资源，也保证了可靠性。

### 二、不同的conf配置文件模拟不同的实例 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70]

### 三、嵌入式Broker ###

![在这里插入图片描述][20210404232311371.png]  
pom.xml  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 1]  
EmbedBroker

package com.atguigu.activemq.embed;
    
    import org.apache.activemq.broker.BrokerService;
    
    /** * @ClassName EmbedBroker * @Description TODO * @Author DuanYueFeng * @Version 1.0 **/
    public class EmbedBroker { 
        public static void main(String[] args) throws Exception { 
            BrokerService brokerService = new BrokerService();
            brokerService.setUseJmx(true);
            brokerService.addConnector("tcp://localhost:61616");
            brokerService.start();
        }
    }

### Spring整合ActiveMQ ###

### 一、maven修改，需要添加spring支持的jms的包 ###

pom.xml

<?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>
    
        <groupId>com.rocket</groupId>
        <artifactId>test_activemq</artifactId>
        <version>1.0-SNAPSHOT</version>
        <properties>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <maven.compiler.source>1.8</maven.compiler.source>
            <maven.compiler.targer>1.8</maven.compiler.targer>
        </properties>
        <dependencies>
            
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-all</artifactId>
                <version>5.15.9</version>
            </dependency>
            <dependency>
                <groupId>org.apache.xbean</groupId>
                <artifactId>xbean-spring</artifactId>
                <version>3.16</version>
            </dependency>
            <dependency>
                <groupId>com.fasterxml.jackson.core</groupId>
                <artifactId>jackson-databind</artifactId>
                <version>2.9.5</version>
            </dependency>
    
    
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-jms</artifactId>
                <version>4.3.23.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-pool</artifactId>
                <version>5.15.9</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-core</artifactId>
                <version>4.3.23.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-context</artifactId>
                <version>4.3.23.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-aop</artifactId>
                <version>4.3.23.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-orm</artifactId>
                <version>4.3.23.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.aspectj</groupId>
                <artifactId>aspectjrt</artifactId>
                <version>1.6.1</version>
            </dependency>
            <dependency>
                <groupId>org.aspectj</groupId>
                <artifactId>aspectjweaver</artifactId>
                <version>1.6.8</version>
            </dependency>
            <dependency>
                <groupId>cglib</groupId>
                <artifactId>cglib</artifactId>
                <version>2.1_2</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-api</artifactId>
                <version>1.7.25</version>
            </dependency>
            <dependency>
                <groupId>ch.qos.logback</groupId>
                <artifactId>logback-classic</artifactId>
                <version>1.2.3</version>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.16.18</version>
            </dependency>
            <dependency>
                <groupId>junit</groupId>
                <artifactId>junit</artifactId>
                <version>4.12</version>
            </dependency>
        </dependencies>
    </project>

### 二、spring配置文件 ###

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
    
            <context:component-scan base-package="com.atguigu.activemq"></context:component-scan>
    
            <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
                    <property name="connectionFactory">
                            <bean class="org.apache.activemq.ActiveMQConnectionFactory">
                                    <property name="brokerURL" value="tcp://10.112.70.211:61616"/>
                            </bean>
                    </property>
                    <property name="maxConnections" value="100"></property>
            </bean>
    
            <bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">
                    <constructor-arg index="0" value="spring-active-queue"/>
            </bean>
    
            
            <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
                    <property name="connectionFactory" ref="jmsFactory" />
                    <property name="defaultDestination" ref="destinationQueue"/>
                    <property name="messageConverter">
                            <bean class="org.springframework.jms.support.converter.SimpleMessageConverter" />
                    </property>
            </bean>
    
    </beans>

### 三、队列 ###

#### 1、生产者 ####

package com.atguigu.activemq.spring;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    import org.springframework.stereotype.Service;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    /** * @ClassName SpringMQ_Consumer * @Description TODO * @Author DuanYueFeng * @Version 1.0 **/
    @Service
    public class SpringMQ_Producer { 
        @Autowired
        private JmsTemplate jmsTemplate;
    
        public static void main(String[] args) { 
            ApplicationContext applicationContext = new ClassPathXmlApplicationContext("applicationContext.xml");
            SpringMQ_Producer producer = (SpringMQ_Producer) applicationContext.getBean("springMQ_Producer");
    
            producer.jmsTemplate.send(new MessageCreator() { 
                @Override
                public Message createMessage(Session session) throws JMSException { 
    
                    TextMessage textMessage = session.createTextMessage("Spring+Active。。。。");
                    return textMessage;
                }
            });
            
        }
    }

#### 2、消费者 ####

package com.atguigu.activemq.spring;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    import org.springframework.stereotype.Service;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    /** * @ClassName SpringMQ_Consumer * @Description TODO * @Author DuanYueFeng * @Version 1.0 **/
    @Service
    public class SpringMQ_Consumer { 
    
        @Autowired
        private JmsTemplate jmsTemplate;
    
        public static void main(String[] args) throws JMSException { 
            ApplicationContext applicationContext = new ClassPathXmlApplicationContext("applicationContext.xml");
            SpringMQ_Consumer consumer = (SpringMQ_Consumer) applicationContext.getBean("springMQ_Consumer");
    // TextMessage receive =(TextMessage) consumer.jmsTemplate.receive();
            String receive =(String) consumer.jmsTemplate.receiveAndConvert();
            System.out.println("消费者："+receive);
        }
    }

### 四、主题 ###

#### 1、Spring配置文件，新增主题topic ####

applicationContext.xml

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
    
            <context:component-scan base-package="com.atguigu.activemq"></context:component-scan>
    
            <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
                    <property name="connectionFactory">
                            <bean class="org.apache.activemq.ActiveMQConnectionFactory">
                                    <property name="brokerURL" value="tcp://10.112.70.211:61616"/>
                            </bean>
                    </property>
                    <property name="maxConnections" value="100"></property>
            </bean>
    
            <bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">
                    <constructor-arg index="0" value="spring-active-topic"/>
            </bean>
    
            
            <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
                    <property name="connectionFactory" ref="jmsFactory" />
                    <property name="defaultDestination" ref="destinationTopic"/>
                    <property name="messageConverter">
                            <bean class="org.springframework.jms.support.converter.SimpleMessageConverter" />
                    </property>
            </bean>
    
    </beans>

#### 2、生产者 ####

同队列代码不变

#### 3、消费者 ####

同队列代码不变

### 五、在spring中里边实现消费者不启动，直接通过配置监听完成 ###

#### 1、spring配置文件 ####

applicationContext.xml

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
    
            <context:component-scan base-package="com.atguigu.activemq"></context:component-scan>
    
            <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
                    <property name="connectionFactory">
                            <bean class="org.apache.activemq.ActiveMQConnectionFactory">
                                    <property name="brokerURL" value="tcp://10.112.70.211:61616"/>
                            </bean>
                    </property>
                    <property name="maxConnections" value="100"></property>
            </bean>
    
            <bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">
                    <constructor-arg index="0" value="spring-active-queue"/>
            </bean>
            <bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">
                    <constructor-arg index="0" value="spring-active-topic"/>
            </bean>
    
            
            <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
                    <property name="connectionFactory" ref="jmsFactory" />
                    <property name="defaultDestination" ref="destinationTopic"/>
                    <property name="messageConverter">
                            <bean class="org.springframework.jms.support.converter.SimpleMessageConverter" />
                    </property>
            </bean>
    
            <bean id="queueListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
                    <property name="connectionFactory" ref="jmsFactory" />
                    <property name="destination" ref="destinationTopic" />
                    <property name="messageListener" ref="myMessageListener" />
            </bean>
    </beans>

#### 2、需要写一个类来实现监听 ####

package com.atguigu.activemq.spring;
    
    import org.springframework.stereotype.Service;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    /** * @ClassName MyMessageListener * @Description TODO * @Author DuanYueFeng * @Version 1.0 **/
    @Service
    public class MyMessageListener implements MessageListener { 
        @Override
        public void onMessage(Message message) { 
            if (null != message && message instanceof TextMessage){ 
                TextMessage textMessage = (TextMessage) message;
                try { 
                    System.out.println("spring消费者接收到消息："+textMessage.getText());
                } catch (JMSException e) { 
                    e.printStackTrace();
                }
            }
        }
    }

#### 3、只需要启动生产者，不需要启动消费者，自动完成监听记录 ####

### SpringBoot整合ActiveMQ ###

### 一、队列 ###

#### 1、队列生产者 ####

##### ①、新建maven工程并设置包名类名 #####

##### ②、pom.xml #####

##### ③、3.application.yaml #####

server:
      port: 7777
    spring:
      activemq:
        broker-url: tcp://10.112.70.211:61616
        user: admin
        password: admin
      jms:
        pub-sub-domain: false
    
    myqueue: boot-activemq-queue

##### ④、配置bean #####

package com.rocket.springboot_activemq.config;
    
    import org.apache.activemq.command.ActiveMQQueue;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.stereotype.Component;
    
    import javax.jms.Queue;
    
    /** * @ClassName ConfigBean * @Description TODO * @Author DuanYueFeng * @Version 1.0 **/
    @Component
    @EnableJms
    public class ConfigBean { 
      @Value("${myqueue}")
        private String myQueue;
    
        @Bean
        public Queue queue(){ 
            return new ActiveMQQueue(myQueue);
        }
    }

类似Spring框架的applicationContext.xml文件

##### ⑤、Queue\_Produce #####

package com.rocket.springboot_activemq.produce;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsMessagingTemplate;
    import org.springframework.stereotype.Component;
    
    import javax.jms.Queue;
    import java.util.UUID;
    
    /** * @ClassName producer * @Description TODO * @Author DuanYueFeng * @Version 1.0 **/
    @Component
    public class Queue_Produce { 
        @Autowired
        private JmsMessagingTemplate jmsMessagingTemplate;
        @Autowired
        private Queue queue;
    
        public void produceMsg(){ 
            jmsMessagingTemplate.convertAndSend(queue,"******:"+ UUID.randomUUID().toString().substring(0,6));
        }
    }

##### ⑥、主启动类MainApp\_Produce #####

##### ⑦、测试单元 #####

##### ⑧、新需求： #####

要求每隔3秒往MQ推送消息，以下定时发送case,案例修改：  
修改Queue\_Produce新增定时投递方法

@Scheduled(fixedDelay = 3000)
    public void produceMsgScheduled(){ 
        jmsMessagingTemplate.convertAndSend(queue,"******Scheduled:"+ UUID.randomUUID().toString().substring(0,6));
        System.out.println("发送消息。。。");
    }

修改主启动类

@EnableScheduling

直接开始主启动类，间隔发消息OK

#### 2、队列消费者 ####

##### ①、新建maven工程并设置包名类名 #####

##### ②、pom.xml #####

##### ③、application.yml #####

server:
      port: 8888
    spring:
      activemq:
        broker-url: tcp://10.112.70.211:61616
        user: admin
        password: admin
      jms:
        pub-sub-domain: false
    
    myqueue: boot-activemq-queue

##### ④、Queue\_Consumer #####

package com.rocket.boot_mq_consumer.consumer;
    
    import org.springframework.jms.annotation.JmsListener;
    import org.springframework.stereotype.Service;
    
    import javax.jms.JMSException;
    import javax.jms.TextMessage;
    
    /** * @ClassName QueueConsumer * @Description TODO * @Author DuanYueFeng * @Version 1.0 **/
    @Service
    public class QueueConsumer { 
    
        @JmsListener(destination = "${myqueue}")
        public void receive(TextMessage textMessage) throws JMSException { 
            System.out.println("消费者接受消息："+textMessage.getText());
        }
    }

##### ⑤、主启动类 #####

package com.rocket.boot_mq_consumer;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    @SpringBootApplication
    public class BootMqConsumerApplication { 
    
        public static void main(String[] args) { 
            SpringApplication.run(BootMqConsumerApplication.class, args);
        }
    
    }

### 二、发布订阅 ###

#### 1、Topic生产者 ####

##### ①、新建maven工程并设置包名类名 #####

##### ②、pom.xml #####

##### ③、application.yaml #####

server:
      port: 7777
    spring:
      activemq:
        broker-url: tcp://10.112.70.211:61616
        user: admin
        password: admin
      jms:
        pub-sub-domain: true
    
    mytopic: boot-activemq-topic

##### ④、配置bean #####

package com.rocket.boot_mq_topic.config;
    
    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.Queue;
    import javax.jms.Topic;
    
    /** * @ClassName ConfigBean * @Description TODO * @Author DuanYueFeng * @Version 1.0 **/
    @Component
    @EnableJms
    public class ConfigBean { 
        @Value("${mytopic}")
        private String myTopic;
    
        @Bean
        public Topic queue(){ 
            return new ActiveMQTopic(myTopic);
        }
    }

##### ⑤、Topic\_Produce #####

package com.rocket.boot_mq_topic.produce;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsMessagingTemplate;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Component;
    
    import javax.jms.Queue;
    import javax.jms.Topic;
    import java.util.UUID;
    
    /** * @ClassName producer * @Description TODO * @Author DuanYueFeng * @Version 1.0 **/
    @Component
    public class Topic_Produce { 
        @Autowired
        private JmsMessagingTemplate jmsMessagingTemplate;
        @Autowired
        private Topic topic;
    
        public void produceMsg(){ 
            jmsMessagingTemplate.convertAndSend(topic,"Topic******:"+ UUID.randomUUID().toString().substring(0,6));
        }
        @Scheduled(fixedDelay = 3000)
        public void produceMsgScheduled(){ 
            jmsMessagingTemplate.convertAndSend(topic,"Topic******Scheduled:"+ UUID.randomUUID().toString().substring(0,6));
            System.out.println("发送消息。。。");
        }
    }

##### ⑥、主启动类MainApp\_Produce #####

package com.rocket.boot_mq_topic;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.scheduling.annotation.EnableScheduling;
    
    @SpringBootApplication
    @EnableScheduling
    public class BootMqTopicApplication { 
    
        public static void main(String[] args) { 
            SpringApplication.run(BootMqTopicApplication.class, args);
        }
    
    }

#### 2、Topic消费者 ####

##### ①、新建maven工程并设置包名类名 #####

##### ②、pom.xml #####

##### ③、application.yml #####

server:
      port: 5555
    spring:
      activemq:
        broker-url: tcp://10.112.70.211:61616
        user: admin
        password: admin
      jms:
        pub-sub-domain: true
    
    myqueue: boot-activemq-topic

##### ④、Topic\_Consumer #####

package com.rocket.boot_mq_topic_consumer.consumer;
    
    import org.springframework.jms.annotation.JmsListener;
    import org.springframework.stereotype.Service;
    
    import javax.jms.JMSException;
    import javax.jms.TextMessage;
    
    /** * @ClassName QueueConsumer * @Description TODO * @Author DuanYueFeng * @Version 1.0 **/
    @Service
    public class TopicConsumer { 
    
        @JmsListener(destination = "${myqueue}")
        public void receive(TextMessage textMessage) throws JMSException { 
            System.out.println("消费者接受消息："+textMessage.getText());
        }
    }

##### ⑤、主启动类 #####

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70]: /images/20221022/cc1fb5e91d82456a9c76a334cd828b88.png
[20210404232311371.png]: /images/20221022/29580f2517594d3f88c7e23794ff0136.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 1]: /images/20221022/23dbd407e04d42e09b4fd7802d252aba.png