【开发技巧】-- SpringBoot快速整合ActiveMQ

墨蓝 2023-02-18 11:56 56阅读 0赞

### 1.1 什么是ActiveMQ？ ###

ActiveMQ是由Apache开源组织维护的一个消息中间件项目，它是基于JMS规范开发的一个消息中间件，它是由Java语言编写的。  
![在这里插入图片描述][20200617175600771.png]

### 1.2 为什么要使用ActiveMQ? ###

在分布式/微服务的系统中，有一部分业务是需要跨越多个服务进行执行的，所以会设计到服务与服务之间的调用，因为要跨越多个服务所以业务处理速度上会有些慢，而如果整合ActiveMQ的话可以将需要执行的业务信息放入队列中，这样服务可以快速响应。通过整合ActiveMQ可以实现：`业务的解耦`、`流量的削峰`、`消息的异步发送`。

### 1.3 在SpringBoot中如何使用ActiveMQ? ###

众所周知SpringBoot为我们提供了大量的`Stater启动器`用以简化我们的开发，所以SpringBoot整合ActiveMQ只需要引入指定的Stater就能够快速的集成使用。

#### 1.3.1 在pom文件中，引入`spring-boot-starter-activemq`与`activemq-pool`集成ActiveMQ ####

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-activemq</artifactId>
        <version>2.3.0.RELEASE</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-pool</artifactId>
        <version>5.15.11</version>
    </dependency>

#### 1.3.2 编写`ActiveMQUtil`工具类，用于简化ActiveMQ的消息发送操作 ####

package com.muzili.mall.utils;
    
    import com.muzili.mall.constants.CommonConstant;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.activemq.ScheduledMessage;
    import org.apache.activemq.pool.PooledConnectionFactory;
    
    import javax.jms.*;
    import java.util.HashMap;
    
    @Slf4j
    public class ActiveMQUtil { 
    
        PooledConnectionFactory pooledConnectionFactory=null;
    
        public ConnectionFactory init(String brokerUrl){ 
            //构建activemq连接工厂对象
            ActiveMQConnectionFactory factory=new ActiveMQConnectionFactory(brokerUrl);
    
            //将连接工厂加入连接池中
            pooledConnectionFactory=new PooledConnectionFactory(factory);
            //设置出现异常时重新连接
            pooledConnectionFactory.setReconnectOnException(true);
            //设置最大连接数为5
            pooledConnectionFactory.setMaxConnections(5);
            //设置连接超时时间
            pooledConnectionFactory.setExpiryTimeout(10000);
    
            return pooledConnectionFactory;
        }
    
        public ConnectionFactory getConnectionFactory(){ 
            return pooledConnectionFactory;
        }
    
        public Connection getConnection() throws JMSException { 
            return this.getConnectionFactory().createConnection();
        }
    
        /** * 发送普通的mapMessage * @param queueName * @param msgMap * @param connection * @return 是否发送成功 */
        public boolean sendMapMessage(String queueName,HashMap<String,Object> msgMap,Connection connection){ 
            return this.sendDefaultMapMessage(queueName,connection,null,msgMap);
        }
    
        /** * 发送延时mapMessage * @param queueName * @param msgMap * @param count * @param connection * @return 是否发送成功 */
        public boolean sendDelayMapMessage(String queueName,HashMap<String,Object> msgMap,Integer count,Connection connection){ 
            return this.sendDefaultMapMessage(queueName,connection,count,msgMap);
        }
    
    
        private boolean sendDefaultMapMessage(String queueName, Connection connection, Integer count, HashMap<String,Object> msgMap){ 
    
            boolean flag=false;
    
            Session session=null;
    
            //创建会话对象
            try { 
                session=connection.createSession(true,Session.SESSION_TRANSACTED);
    
                //构建队列信息
                Queue queue=session.createQueue(queueName);
    
                //构造一个消息对象
                MapMessage mapMessage=session.createMapMessage();
    
                //如果count不为null表示他是一个延时消息
                if (count!=null){ 
                    if (count< CommonConstant.DELAY_MSG_MAX_RESEND_COUNT){ 
                        //延时消息重发次数如果小于最大重试次数 那么 本次延时时间=当前次数*最小等待时间
                        Integer delayTime=count*CommonConstant.DELAY_MSG_QUEUE_INIT_WAIT_TIME;
    
                        log.info("延时消息将在{}ms后被推送！",delayTime);
    
                        mapMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY,delayTime);
                    }else{ 
                        log.info("延时消息将在24小时后被推送！");
    
                        mapMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY,CommonConstant.DELAY_MSG_QUEUE_INIT_WAIT_TIME*60*24);
                    }
                }
    
                //将待发送的消息设置到mapMessage对象中
                msgMap.forEach((k,v)->{ 
                    try { 
                        mapMessage.setObject(k,v);
                    } catch (JMSException e) { 
                        e.printStackTrace();
                    }
                });
    
                //创建消息的生产者
                MessageProducer producer=session.createProducer(queue);
    
                //发送消息
                producer.send(mapMessage);
    
                //关闭消息生产者
                producer.close();
    
                flag=true;
    
            } catch (JMSException e) { 
    
                if (session != null) { 
                    try { 
                        session.close();
                    } catch (JMSException ex) { 
                        ex.printStackTrace();
                    }
                }
    
                e.printStackTrace();
            }finally { 
                try { 
                    //释放资源
                    if (session != null) { 
                        session.close();
                    }
    
                    if (connection != null) { 
                        connection.close();
                    }
                } catch (JMSException e) { 
                    e.printStackTrace();
                }
            }
    
            return flag;
        }
    
    
    }

#### 1.3.3 编写`ActiveMQConfig`配置类 ####

package com.muzili.mall.payment.config;
    
    import com.muzili.mall.utils.ActiveMQUtil;
    import org.apache.activemq.ActiveMQConnectionFactory;
    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 javax.jms.Session;
    
    @Configuration
    public class ActiveMQConfig { 
    
        @Value("${spring.activemq.broker-url}")
        private String brokerURL;
    
        @Value("${activemq.listener.enable}")
        private boolean listenerEnable;
    
        @Bean
        public ActiveMQUtil getActiveMQ(){ 
    
            if (brokerURL.equals("disabled")){ 
                return null;
            }
    
            ActiveMQUtil activeMQUtil=new ActiveMQUtil();
            activeMQUtil.init(brokerURL);
    
            return activeMQUtil;
        }
    
    
        public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(ActiveMQConnectionFactory activeMQConnectionFactory){ 
            DefaultJmsListenerContainerFactory factory=new DefaultJmsListenerContainerFactory();
    
            //如果配置中没有设置这不开启
            if (!listenerEnable){ 
                return null;
            }
    
            //设置连接工厂
            factory.setConnectionFactory(activeMQConnectionFactory);
            //设置并发数
            factory.setConcurrency("5");
            //设置重连间隔时间
            factory.setRecoveryInterval(5000L);
            //设置签收模式
            factory.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
    
            return factory;
        }
    
        /** * 注入activemq连接工厂对象 * @return */
        @Bean
        public ActiveMQConnectionFactory activeMQConnectionFactory(){ 
            ActiveMQConnectionFactory factory=new ActiveMQConnectionFactory(brokerURL);
    
            return factory;
        }
    }

#### 1.3.4 在SpringBoot配置文件中编写相关配置信息 ####

#activemq相关配置信息
    spring.activemq.broker-url=tcp://localhost:61616
    activemq.listener.enable=true

#### 1.3.5 编码发送消息到ActiveMQ ####

@Autowired
    private ActiveMQUtil activeMQUtil;
    
    public void testSendMessage(){ 
    	 HashMap<String,Object> msgMap=new HashMap<>();
    
         msgMap.put("orderSn",fda12f31fda234520);
         msgMap.put("payType", 1);
    
         try { 
             activeMQUtil.sendMapMessage("queueName",msgMap,activeMQUtil.getConnection());
         } catch (JMSException e) { 
             e.printStackTrace();
         }
    }

[20200617175600771.png]: https://img-blog.csdnimg.cn/20200617175600771.png