springboot项目中应用消息队列ActiveMQ

朴灿烈づ我的快乐病毒、 2022-12-26 09:15 117阅读 0赞

springcloud 中应用activeMq的步骤：

1.pom文件引入依赖

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

2.yml配置文件

spring:
      activemq:
        broker-url: tcp://127.0.0.1:61616
        user: admin
        password: admin
        pool:
          enabled: true
          max-connections: 100

3.启动类中增加@EnableJms注解

4.生产者端

*  4.1 配置类

import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.activemq.command.ActiveMQQueue;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import javax.jms.Queue;
    
    
    @Configuration
    public class ActiveMQConfig {
    
        @Value("${spring.activemq.broker-url}")
        private String brokerUrl;
    
        //创建消息队列，并设置名称
        @Bean
        public Queue queue() {
            return new ActiveMQQueue("ActiveMQQueue");
        }
    
        // 连接消息队列
        @Bean
        public ActiveMQConnectionFactory connectionFactory(){
            return new ActiveMQConnectionFactory(brokerUrl);
        }
    }

*  4.2  创建生产者 生产消息

import net.sf.json.JSONObject;
    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 org.springframework.transaction.annotation.Transactional;
    
    import javax.jms.Queue;
    import java.util.List;
    
    @Component
    public class Produce {
    
    
        @Autowired
        private Queue queue;
    
        @Autowired
        JmsMessagingTemplate jmsMessagingTemplate;
    
       
        public void task(){
            //获取要发送的消息内容，指定消息队列，发送消息
    //event 为要发送消息的内容，可以从数据库或者redis等获取并组装数据
                jmsMessagingTemplate.convertAndSend(queue,JSONObject.fromObject(event).toString());
     
    
        }
    
    }

5.消费端

5.1配置类

import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.activemq.RedeliveryPolicy;
    import org.apache.activemq.command.ActiveMQQueue;
    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;
    
    
    /**
     * 消息消费端 的配置
     */
    @Configuration
    public class ActiveMQConfig {
    
        @Value("${spring.activemq.broker-url}")
        private String brokerUrl;
    
        /**
         * 连接工厂
         * @param redeliveryPolicy
         * @return
         */
        @Bean
        public ActiveMQConnectionFactory connectionFactory(RedeliveryPolicy redeliveryPolicy){
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory("admin","admin",brokerUrl);
            activeMQConnectionFactory.setRedeliveryPolicy(redeliveryPolicy);
            return activeMQConnectionFactory;
        }
    
        /**
         * 重发配置
         * @return
         */
        @Bean
        public RedeliveryPolicy redeliveryPolicy(){
            RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
            return redeliveryPolicy;
        }
    
        /**
         * 设置消息队列 确认机制
         * @param activeMQConnectionFactory
         * @return
         */
        @Bean
        public JmsListenerContainerFactory jmsListenerContainerFactory(ActiveMQConnectionFactory activeMQConnectionFactory){
            DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
            bean.setConnectionFactory(activeMQConnectionFactory);
            // 1: 自动确认，2： 客户端手动确认，3：自动批量确认，4 事务提交并确认。
            bean.setSessionAcknowledgeMode(2);
            return bean;
        }
    
    }

*  5.2 消费者消费消息
    
        import net.sf.json.JSONObject;
        import org.springframework.beans.factory.annotation.Autowired;
        import org.springframework.jms.annotation.JmsListener;
        import org.springframework.stereotype.Component;
        
        import javax.jms.JMSException;
        import javax.jms.Session;
        import javax.jms.TextMessage;
        
        
        @Component
        public class ConsumerQueue {
        
        
            @JmsListener(destination = "ActiveMQQueue",containerFactory = "jmsListenerContainerFactory")
            public void receive(TextMessage textMessage, Session session) throws JMSException {
                try {
                    System.out.println("收到的消息："+textMessage.getText());
                    String content = textMessage.getText();
        // 得到消息，开始处理自己相关业务
                   ………………
                    // 业务完成，确认消息 消费成功
                    textMessage.acknowledge();
                }catch (Exception e){
                    // 回滚消息
                    e.printStackTrace();
        //            e.getMessage(); // 放到log中。
                    System.out.println("异常了");
                    session.recover();
                }
        
            }
        
            /**
             * 补偿 处理（人工，脚本）。自己根据自己情况。
             * @param text
             */
            @JmsListener(destination = "DLQ.ActiveMQQueue")
            public void receive2(String text){
                System.out.println("死信队列:"+text);
            }
        }