如何使用Spring Boot JMS监听主题

Question

我正在尝试使用以下代码段来监听主题。但是它默认只能监听队列。在这种情况下没有xml配置。我完全依赖注释。此外，我完全依赖Spring Boot提供的自动配置。我不知道如何在JmsListener中设置目标类型为主题。请Spring JMS专家帮忙。

    @Component
    public class MyTopicListener {

        @JmsListener(destination = "${trans.alert.topic}")
        public void receiveMessage(TransactionAlert alert) {
            logger.info("AlertSubscriberEmail :: Sending Email => <" + alert + ">");
        }
    }

- zikzakjack

Answer 1

被标记为正确的答案几乎是正确的。但它仍然不起作用，因为：

factory.setPubSubDomain(true) 

必须在以下内容之后出现：

configurer.configure(factory, connectionFactory);

否则，如果在配置默认值时将pubSubDomain标志设置为true，则该标志将丢失，并且该工厂实例仍将使用队列而不是主题。

否则，在配置默认值时将pubSubDomain标志设置为true时，该标志将丢失，因此该工厂实例仍将使用队列而不是主题。

- dleerob

---

谢谢你。这让我疯狂了几个小时。 - Assafs

Answer 2

我刚刚从https://github.com/spring-guides/gs-messaging-jms/获取了完整的 Spring Boot 示例。

该示例用于发送和接收队列中的消息。要将其更改为主题，您需要在工厂实例中设置 Pub-Sub 属性。

import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.config.JmsListenerContainerFactory;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.support.converter.MappingJackson2MessageConverter;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.converter.MessageType;

import javax.jms.ConnectionFactory;

@SpringBootApplication
@EnableJms
public class JmsSampleApplication {

public void registerBeans(ConfigurableApplicationContext context ){
    BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition(JmsTemplate.class);
    CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();

    builder.addPropertyValue("connectionFactory", cachingConnectionFactory);      // set property value
    DefaultListableBeanFactory factory = (DefaultListableBeanFactory) context.getAutowireCapableBeanFactory();
    factory.registerBeanDefinition("jmsTemplateName", builder.getBeanDefinition());
}

@Bean
public JmsListenerContainerFactory<?> topicListenerFactory(ConnectionFactory connectionFactory,
                                                DefaultJmsListenerContainerFactoryConfigurer configurer) {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    factory.setPubSubDomain(true);
    // This provides all boot's default to this factory, including the message converter
    configurer.configure(factory, connectionFactory);
    // You could still override some of Boot's default if necessary.
    return factory;
}

@Bean
public JmsListenerContainerFactory<?> queueListenerFactory(ConnectionFactory connectionFactory,
                                                           DefaultJmsListenerContainerFactoryConfigurer configurer) {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    //factory.setPubSubDomain(true);
    // This provides all boot's default to this factory, including the message converter
    configurer.configure(factory, connectionFactory);
    return factory;
}

@Bean // Serialize message content to json using TextMessage
public MessageConverter jacksonJmsMessageConverter() {
    MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
    converter.setTargetType(MessageType.TEXT);
    converter.setTypeIdPropertyName("_type");
    return converter;
}
public static void main(String[] args) {
    ConfigurableApplicationContext context = SpringApplication.run(JmsSampleApplication.class, args);

    JmsTemplate jmsTemplate = context.getBean(JmsTemplate.class);

    // Send a message with a POJO - the template reuse the message converter
    System.out.println("Sending an email message.");
    jmsTemplate.convertAndSend("mailbox.topic", new Email("[e-mail]", "Hello"));
    jmsTemplate.convertAndSend("mailbox.queue", new Email("[e-mail]", "Hello"));

    }
}

听众

package org.springboot.jms;

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

/**
 * Created by RGOVIND on 10/20/2016.
 */
@Component
public class HelloTopicListener {

    @JmsListener(destination = "mailbox.topic", containerFactory = "topicListenerFactory")
    public void receiveTopicMessage(Email email) {
        System.out.println("Received <" + email + ">");
    }

    @JmsListener(destination = "mailbox.queue", containerFactory = "queueListenerFactory")
    public void receiveQueueMessage(Email email) {
        System.out.println("Received <" + email + ">");
    }
}

一旦完成这些步骤，您就可以订阅所需的主题了。

当然，有多种方法可以实现这一点。您可以为不同的jmsTemplates拥有一个bean映射，每个映射都可以根据队列或主题在需要时使用。模板和bean可以在您选择的方法中实例化，例如在此SO问题中讨论的方式。希望它能帮到您。

- Ramachandran.A.G

---

当我尝试使用JmsListenerContainerFactory bean或在application.properties中设置此属性spring.jms.pub-sub-domain=true时，我的所有其他队列发送器和监听器的代码开始表现得好像它们正在读写主题而不是队列。是否有解决方案可以使队列和主题的监听器共存？ - zikzakjack

你可以随时选择拥有多个工厂。在代码的情况下，你可以选择同时拥有队列和主题。问题在于当你选择接收时，你必须选择工厂。在这种情况下，你可以拥有两个工厂，一个从主题获取，另一个从队列获取。我也编辑了我的回答。 - Ramachandran.A.G

我尝试使用多个工厂，但即使如此，创建的主题仍然像队列一样运行，也就是说，在创建多个侦听器时，消息会在它们之间分布。实际上，每个订阅者都应该收到所有的消息。 - Nishant Lakhara

上述示例无法正常工作，这就是为什么你说它仍然像队列而不是主题。原因是因为"factory.setPubSubDomain(true);"必须在"configurer.configure(factory, connectionFactory);"之后执行。 - dleerob

Answer 3

在Spring Boot的Application.properties文件中，尝试设置以下属性：

spring.jms.pub-sub-domain=true

然后，将此属性用于你正在使用的容器工厂来监听主题。

- Aditya K

---

实际上，Spring Boot 自动配置的默认容器工厂将自动使用该标志。 - Stephane Nicoll

我怀疑使用这个属性不会使所有可用的监听器都监听主题。我正在寻找一种在同一个项目中为队列和主题拥有分开的监听器的方法。 - zikzakjack

你不能使用相同的connectionFactory来监听队列和主题。你的监听器将从一个connectionFactory派生出来，该connectionFactory将被配置为监听点对点（Queue）或发布-订阅（Topic）。因此，需要有两个connectionFactories，一个用于主题，一个用于队列，然后根据需要使用它们。pub-sub-domain的默认设置为false，这意味着默认情况下它将监听队列。 - Aditya K

设置属性会覆盖所有创建的pubsubdomain标志工厂，因此所有侦听器都会根据设置的标志值表现为队列或主题。 - Nishant Lakhara