SpringBoot（14） - 消息（2） - JMS（2）发送Queue和Topic

Myth丶恋晨 2022-04-22 13:56 177阅读 0赞

参考：[https://docs.spring.io/spring-boot/docs/1.5.17.RELEASE/reference/htmlsingle/\#boot-features-using-jms-receiving][https_docs.spring.io_spring-boot_docs_1.5.17.RELEASE_reference_htmlsingle_boot-features-using-jms-receiving]

org.springframework.jms.annotation.EnableJms

[https://blog.csdn.net/liuchuanhong1/article/details/72726337][https_blog.csdn.net_liuchuanhong1_article_details_72726337]

同时发送Queue和Topic示例（具体参考官方文档和@EnableJms相关源码）：

1. 添加依赖

<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. application.yml配置

spring:
      activemq:
    #    broker-url: tcp://activemq01.momo.com:61616
        user: admin
        password: admin
        pool:
          enabled: true
          max-connections: 50
          expiry-timeout: 10000
          idle-timeout: 30000
    
    # 测试使用的quue和topic      
    mq: 
      queue: 
        test: test.queue
      topic: 
        test: test.topic

3. ActiveMQ配置

根据官方文档所说的，通过DefaultJmsListenerContainerFactory可以创建自定义的JmsListenerContainerFactory实例，之后在@JmsListener注解中通过containerFactory属性引用它。

在这里主要是配置是否启用发布订阅模式，以用来可以同时接收Queue和Topic消息。具体配置可以参考@EnableJms和DefaultJmsListenerContainerFactory源码。

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 javax.jms.ConnectionFactory;
    
    @Configuration
    @EnableJms
    public class ActiveMQConfig {
        @Bean("queueListenerFactory")
        public JmsListenerContainerFactory<?> queueListenerFactory(ConnectionFactory connectionFactory) {
            DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
            factory.setPubSubDomain(false);
            factory.setConcurrency("5");
            return factory;
        }
    
        @Bean("topicListenerFactory")
        public JmsListenerContainerFactory<?> topicListenerFactory(ConnectionFactory connectionFactory) {
            DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
            factory.setPubSubDomain(true);
            return factory;
        }
    }

4. 定义发送消息工具类

@Component
    public class JmsMessageSender {
        private static final Logger LOG = LoggerFactory.getLogger(JmsMessageSender.class);
        
        @Autowired
        private JmsMessagingTemplate jmsTemplate;
    
        public void sendTextMessage(Destination destination, String data) throws JmxException {
            jmsTemplate.convertAndSend(destination, data);
        }
    
        public void sendJsonMessage(Destination destination, Object data) throws JmxException {
            jmsTemplate.convertAndSend(destination, JSON.toJSONString(data));
        }
    }

5. 队列消息

5.1 发送

@Service
    public class Producer {
        @Value("${mq.queue.test}")
        private String testQueue;
        
        @Autowired
        private JmsMessageSender jmsMessageSender;
        
        private Destination testQueueDestination;
        
        @PostConstruct
        public void init() {
            testQueueDestination = new ActiveMQQueue(testQueue);
        }
        
        public void sendTextMessage(String message) {
            jmsMessageSender.sendTextMessage(testQueueDestination, message);
        }
    }

5.2 接收

@Service
    public class Consumer {
        @JmsListener(destination = "${mq.queue.test}", containerFactory = "queueListenerFactory")
        public void receiveQueue(TextMessage text) throws Exception {
            System.out.println(Thread.currentThread().getName() + " 收到的报文为：" + text.getText());
        }
    }

6. 发布/订阅

6.1 发布

@Service
    public class Publisher {
        @Value("${mq.topic.test}")
        private String testTopic;
        
        @Autowired
        private JmsMessageSender jmsMessageSender;
        
        private Destination testTopicDestination;
        
        @PostConstruct
        public void init() {
            testTopicDestination = new ActiveMQTopic(testTopic);
        }
        
        public void sendTextMessage(String message) {
            jmsMessageSender.sendTextMessage(testTopicDestination, message);
        }
    }

6.2 订阅

@Service
    public class Subscriber {
        @JmsListener(destination = "${mq.topic.test}", containerFactory = "topicListenerFactory")
        public void receiveTopic1(TextMessage text) throws Exception {
            System.out.println("\t" + Thread.currentThread().getName() + " 收到的报文为:" + text.getText());
        }
    
        @JmsListener(destination = "${mq.topic.test}", containerFactory = "topicListenerFactory")
        public void receiveTopic2(TextMessage text) throws Exception {
            System.out.println("\t\t" + Thread.currentThread().getName() + " 收到的报文为:" + text.getText());
        }
    }

7. 测试

@RunWith(SpringRunner.class)
    @SpringBootTest
    public class MessageTests {
        @Autowired
        private Producer producer;
        @Autowired
        private Publisher publisher;
    
        @Test
        public void sendQueueMessage() {
            for (int i = 0; i < 50; i++) {
                producer.sendTextMessage("queue message, i = " + i);
            }
        }
    
        @Test
        public void sendTopicMessage() {
            for (int i = 0; i < 50; i++) {
                publisher.sendTextMessage("topic message, i = " + i);
            }
        }
    }

[https_docs.spring.io_spring-boot_docs_1.5.17.RELEASE_reference_htmlsingle_boot-features-using-jms-receiving]: https://docs.spring.io/spring-boot/docs/1.5.17.RELEASE/reference/htmlsingle/#boot-features-using-jms-receiving
[https_blog.csdn.net_liuchuanhong1_article_details_72726337]: https://blog.csdn.net/liuchuanhong1/article/details/72726337