SpringBoot整合Active MQ,实现队列消息的提供和消费

布满荆棘的人生 2022-11-14 13:12 166阅读 0赞

1，pom.xml文件中添加active MQ的依赖，spring boot框架中已整合了Active MQ的jar包

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

2，application.yml文件中添加MQ配置

spring:
      activemq:
        user: admin
        password: admin
        broker-url: failover:(tcp://192.168.1.1:61616)?randomize=false

3，MQ配置文件ActiveMQConfig.java，负责连接MQ服务端，以及创建队列和主题的消息监听

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 org.springframework.jms.config.JmsListenerContainerFactory;
    
    @Configuration
    public class ActiveMQConfig { 
        @Value("${spring.activemq.user}")
        private String usrName;
    
        @Value("${spring.activemq.password}")
        private  String password;
    
        @Value("${spring.activemq.broker-url}")
        private  String brokerUrl;
    
        // 连接MQ服务端
        @Bean
        public ActiveMQConnectionFactory connectionFactory() { 
            return new ActiveMQConnectionFactory(usrName, password, brokerUrl);
        }
    
        // MQ的队列监听
        @Bean("jmsListenerContainerQueue")
        public JmsListenerContainerFactory<?> jmsListenerContainerQueue(ActiveMQConnectionFactory connectionFactory){ 
            DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
            bean.setConnectionFactory(connectionFactory);
            return bean;
        }
    
        // MQ的主题监听，本文不过多阐述主题的使用
        @Bean("jmsListenerContainerTopic")
        public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ActiveMQConnectionFactory connectionFactory){ 
            DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
            //设置为发布订阅方式, 默认情况下使用的生产消费者方式
            bean.setPubSubDomain(true);
            bean.setConnectionFactory(connectionFactory);
            return bean;
        }
    }

4，监听指定队列的消息并消费 ， MqQueueListener.java

import com.alibaba.fastjson.JSONObject;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.jms.annotation.JmsListener;
    import org.springframework.stereotype.Component;
    
    @Slf4j
    @Component
    public class MqQueueListener { 
    
        // 监听队列名称为queue_name的消息，一旦有数据推送至服务端，则监听器负责消费消息
        @JmsListener(destination = "queue_name", containerFactory = "jmsListenerContainerQueue")
        public void receiveMqQueueMessage(String messageStr) { 
    
            JSONObject messageJson = JSONObject.parseObject(messageStr);
    
            // 方法体
        }

5，向MQ服务端指定的队列推送消息，MQQueueProducer.java

import lombok.extern.slf4j.Slf4j;
    import org.apache.activemq.command.ActiveMQQueue;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsMessagingTemplate;
    import org.springframework.stereotype.Service;
    
    import javax.jms.Destination;
    import javax.jms.Message;
    import javax.jms.TextMessage;
    
    /** * @Description:队列消息生产者 */
    @Slf4j
    @Service("queueProducer")
    public class MQQueueProducer { 
        @Autowired
        private JmsMessagingTemplate jmsTemplate;
    
        // desName为队列名称，message为所推送的消息
        public void sendMessage(String desName, final String message){ 
            Destination destination = new ActiveMQQueue(desName);
            jmsTemplate.convertAndSend(destination, message);
        }
    }