Springboot集成activemq（发布订阅模式topic）

╰半橙微兮° 2022-01-31 05:11 370阅读 0赞

**一、docker安装activemq**

1、查询镜像

> docker search activemq

2、拉取镜像

> docker pull webcenter/activemq

3、查询是否拉取成功

> docker images

4、运行activemq镜像

> docker run -d --name myactivemq -p 61617:61616 -p 8162:8161 webcenter/activemq:latest

注意：

61617为项目中tcp链接的端口

8162位浏览器访问的接口

5、查看运行

> dockr logs -tf --tail 2000 容器id

6、访问，使用192.168.98.105:61617 账号密码：admin/admin

![20190515180500294.png][]

**二、Springboot集成activemqmq**

1.  添加依赖

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-activemq</artifactId>
        <version>1.5.6.RELEASE</version>
    </dependency>
    
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-jms</artifactId>
    </dependency>

2、配置application.yml，重点注意一下**spring.jms.pub-sub-domain=true**，因为activemq默认采用点到点的通信queue，所以需要使用该配置指定使用topic进行通信

spring:
      activemq:
        user: admin
        password: admin
        broker-url: tcp://192.168.98.105:61617
        # 默认代理URL是否应该在内存中。如果指定了显式代理，则忽略此值。
        in-memory: true
        # true表示使用连接池
        pool:
          enabled: true
          # 连接池最大连接数
          max-connections: 5
        # mq默认的提供的是queue, 若要topic需要如下配置
        non-blocking-redelivery: false
      jms:
        pub-sub-domain: true
    
    activeMq:
      acs:
        topicName: taskNodeLog

3、添加config配置信息

import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.activemq.command.ActiveMQTopic;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.jms.annotation.EnableJms;
    import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
    import org.springframework.jms.config.JmsListenerContainerFactory;
    import org.springframework.jms.core.JmsMessagingTemplate;
    import org.springframework.jms.core.JmsTemplate;
    
    import javax.jms.ConnectionFactory;
    import javax.jms.Topic;
    
    @Configuration
    public class ActiveMqConfig {
    
        @Value("${activeMq.acs.topicName}")
        private String topic;
    
        @Value("${spring.activemq.broker-url}")
        private String brokeUrl;
    
        @Value("${spring.activemq.user}")
        private String username;
    
        @Value("${spring.activemq.password}")
        private String password;
    
        /**
         * 发布订阅模式
         * @return
         */
        @Bean
        public Topic topic() {
    
            return new ActiveMQTopic(topic);
        }
    
        @Bean
        public ConnectionFactory connectionFactory(){
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
            connectionFactory.setBrokerURL(brokeUrl);
            connectionFactory.setUserName(username);
            connectionFactory.setPassword(password);
            return connectionFactory;
        }
    
        @Bean("jmsTemplate")
        public JmsTemplate genJmsTemplate(){
            return new JmsTemplate(connectionFactory());
    
        }
    
        @Bean("jmsMessagingTemplate")
        public JmsMessagingTemplate jmsMessageTemplate(){
            return new JmsMessagingTemplate(connectionFactory());
        }
    
    }

4、添加生产者发送信息的方法

import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.annotation.JmsListener;
    import org.springframework.jms.core.JmsMessagingTemplate;
    import org.springframework.stereotype.Component;
    import javax.jms.Destination;
    
    
    @Component
    @Slf4j
    public class ActiveMqProducer {
    
        @Autowired
        private JmsMessagingTemplate jmsMessagingTemplate;
    
        /**
         * 把接收到的消息和消息模式放到了队列里
         * 或者主题里就是queue或topic里，然后只要写
         * 消费者加上@JmsListener监听队列消息就可以自动获取
         * @param destination
         * @param message
         */
        public void sendMessage(Destination destination, String message) {
            jmsMessagingTemplate.convertAndSend(destination, message);
        }
    
        /**
         * 这里的第二个方法加上了监听注解就可以收到消费者返回的信息，前提消费者要加@SendTo注解，
         * @param message
         */
        @JmsListener(destination = "return-topic")
        public void Message(String message) {
            log.info("生产者收到的返回消息：{}", message );
        }
    }

5、添加消息监听

@Component
    public class ActiveMqListener {
    
        @JmsListener(destination = "${activeMq.acs.topicName}", containerFactory = "jmsListenerContainerTopic")
        @SendTo("return-topic")
        public void receiveTopic(String message) {
            System.out.println("监听topic=============yssTask");
            System.out.println(message);
        }
    
    }

6、添加controller进行消息发送的测试

@RestController
    @RequestMapping("/activemq")
    public class ActiveMqController {
    
        @Autowired
        private ActiveMqProducer activeMqProducer;
    
        @Autowired
        private Topic topic;
    
        @RequestMapping("/topic")
        public String handlerActiveMq(@RequestParam("msg")String msg) {
            activeMqProducer.sendMessage(topic, msg);
            return "ok";
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70][]

注意：

为了防止自产自销，在项目中的监听才用了固定的destination，如果生产中请避免自我消费，更改taskName值即可

@RestController注意

源码地址：[https://gitee.com/SmileLsf/activemq\_demo][https_gitee.com_SmileLsf_activemq_demo]

[20190515180500294.png]: /images/20220131/adf5dd39c301460bb72611978ce04d70.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1Mjc1MjMz_size_16_color_FFFFFF_t_70]: /images/20220131/b2700ec4ce5b4830a3cdbde4737187b8.png
[https_gitee.com_SmileLsf_activemq_demo]: https://gitee.com/SmileLsf/activemq_demo