如何从其他应用程序订阅Spring Boot JMS主题

Question

我有两个应用程序，第一个应用程序启动ActiveMQ代理（https://spring.io/guides/gs/messaging-jms/ ).

在第二个应用程序，我想从第一个应用程序的主题。

我如何在不启动ActiveMQ服务器的情况下执行此操作？

可能的解决办法：

服务器应用程序项目

import java.time.LocalDateTime;

import javax.jms.ConnectionFactory;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.springframework.beans.factory.annotation.Autowired;
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.DefaultJmsListenerContainerFactory;
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 org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.ui.ModelMap;

@SpringBootApplication
@EnableJms
@EnableScheduling
public class JsmServerApplication {

    @Autowired
    JmsTemplate jmsTemplate;

    @Bean
    public BrokerService broker() throws Exception {
        BrokerService ret = new BrokerService();
        ret.addConnector("tcp://0.0.0.0:4444"); // allow remote connections
        ret.setBrokerName("primary-broker");
        ret.setUseJmx(true);
        return ret;
    }

    @Bean
    public ConnectionFactory connectionFactory() {
        return new ActiveMQConnectionFactory("tcp://localhost:4444");
    }

    @Bean
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(jacksonJmsMessageConverter());
        return factory;
    }

    @Bean
    public MessageConverter jacksonJmsMessageConverter() {
        MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
        converter.setTargetType(MessageType.TEXT);
        converter.setTypeIdPropertyName("_type");
        return converter;
    }

    public static void main(String[] args) {
        SpringApplication.run(JsmServerApplication.class, args);
    }

    @Scheduled(cron = "*/5 * * * * ?")
    public void run() {
        ModelMap msg = new ModelMap("now", LocalDateTime.now().toString());
        System.out.println("Sending: " + msg);
        jmsTemplate.convertAndSend("messages", msg);
    }

}

客户端应用程序项目

import javax.jms.ConnectionFactory;

import org.apache.activemq.ActiveMQConnectionFactory;
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.annotation.JmsListener;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.support.converter.MappingJackson2MessageConverter;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.converter.MessageType;
import org.springframework.ui.ModelMap;

@SpringBootApplication
@EnableJms
public class JsmClientApplication {

    @Bean
    public ConnectionFactory connectionFactory() {
        return new ActiveMQConnectionFactory("tcp://localhost:4444");
    }

    @Bean
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(jacksonJmsMessageConverter());
        return factory;
    }

    @Bean
    public MessageConverter jacksonJmsMessageConverter() {
        MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
        converter.setTargetType(MessageType.TEXT);
        converter.setTypeIdPropertyName("_type");
        return converter;
    }

    @JmsListener(destination = "messages", containerFactory = "jmsListenerContainerFactory")
    public void msg(ModelMap msg) {
        System.out.println(msg);
    }

    public static void main(String[] args) {
        SpringApplication.run(JsmClientApplication.class, args);
    }

}

这是正确的方法吗？

用这个解决了：

http://javasampleapproach.com/java-integration/activemq-work-spring-jms-activemq-topic-publisher-subcribers-pattern-using-springboot

Answer 1

有关如何在tcp端口而不是vm://传输端口上侦听的信息，请参阅此答案。

Answer 2

您可以使用Message消费者来使用如下代码所示的数据

public static void main(String[] args) throws JMSException {
    // Getting JMS connection from the server
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
    Connection connection = connectionFactory.createConnection();
    connection.start();

    Session session = connection.createSession(false,
                Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTopic("topic");
    MessageConsumer consumer = session.createConsumer(topic);

    MessageListener listner = new MessageListener() {
        public void onMessage(Message message) {
            try {
                //do operations
           } catch (JMSException e) {

           }
         }
    };

    consumer.setMessageListener(listner);
    connection.close();

}

由于您使用的是ActiveMQConnectionFactory，您可以设置如下代理

BrokerService broker = new BrokerService();
broker.addConnector("tcp://localhost:4444);
broker.setPersistent(false);

ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");

如果您在不使用ActiveMQ方面没有任何限制，那么可以使用Kafka来执行相同的操作。Kafka通过简单的API为您提供了一个高度可扩展的分布式消息总线。

https://kafka.apache.org/quickstart

我不确定限制，但我只是想给你一种Kafka的感觉。然而，上面的代码应该有助于您理解订阅和使用主题消息的概念。