SpringBoot整合ActiveMQ

╰半橙微兮° 2021-11-04 19:32 429阅读 0赞

1.  引入Jar包

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

1.  修改配置文件application.yml

spring:
      activemq:
        broker-url: tcp://192.168.227.131:61616
        #默认的broker-url是否使用缓存.
        in-memory: false
        pool:
          #是否使用jms连接池
          enabled: true
          #连接池中的最大连接数
          max-connections: 10
          #空闲的连接过期时间，默认为30秒
          idle-timeout: 30000

1.  在main启动类上添加@EnableJms注解

@SpringBootApplication
    @EnableJms  //启动消息队列
    public class Application {
        public static void main(String[] args) {
            SpringApplication.run(Application.class, args);
        }
    }

1.  创建Controll类发布queue和topic信息

import org.apache.activemq.command.ActiveMQTopic;
    import org.apache.commons.lang3.RandomUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsMessagingTemplate;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import javax.jms.Destination;
    
    @RestController
    public class ProducerController {
        @Autowired
        private JmsMessagingTemplate jmsMessagingTemplate; //注入springboot封装的工具类
    
        @RequestMapping("send")
        public void send() {
            jmsMessagingTemplate.convertAndSend("pegasus", String.valueOf(RandomUtils.nextInt(1, 10000)));
        }
    
        @RequestMapping("publish")
        public void publish() {
            Destination destination = new ActiveMQTopic("pegasus_topic");
            jmsMessagingTemplate.convertAndSend(destination, "have a test");
        }
    }

1.  创建Service类接受Queue和topic信息

package com.qf.activemq;
    
    import org.springframework.jms.annotation.JmsListener;
    import org.springframework.stereotype.Component;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    @Component
    public class ConsumerService {
        private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    
        @JmsListener(destination = "pegasus")
        public void receiveMessage(String message) {
            System.out.println(String.format("%s:  收到消息 - %s", sdf.format(new Date()), message));
        }
    
        @JmsListener(destination = "pegasus_topic", containerFactory = "myJmsContainerFactory")
        public void subscribe(String message) {
            System.out.println(String.format("%s:  收到订阅消息 - %s", sdf.format(new Date()), message));
        }
    }

1.  接受topic需要配置JmsListenerContainerFactory类

import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.context.annotation.Bean;
    import org.springframework.jms.annotation.EnableJms;
    import org.springframework.jms.config.JmsListenerContainerFactory;
    import org.springframework.jms.config.SimpleJmsListenerContainerFactory;
    
    import javax.jms.ConnectionFactory;
    
    @SpringBootApplication
    @EnableJms
    public class Application {
        public static void main(String[] args) {        
            SpringApplication.run(Application.class, args);
        }
    
        @Bean
        public JmsListenerContainerFactory<?> myJmsContainerFactory(ConnectionFactory connectionFactory) {
            SimpleJmsListenerContainerFactory factory = new SimpleJmsListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
            factory.setPubSubDomain(true);
            return factory;
        }
    }