无法使用ActiveMQ同时向两个订户发布主题消息

Question

我提到了SpringBoot应用程序发布和读取ActiveMQ主题，使用ActiveMQ发布一个主题。我已经创建了两个从主题中读取消息的接收器微服务。我还创建了rest endpoint来发布主题。但是，我必须执行这个restendpoint两次，以便为两个接收者发布消息。restendpoint的第一次执行将向Receiver1发送消息。restendpoint的第二次执行将向Receiver2发送消息

因此，两个接收器不能同时读取主题
这是我的代码。

PublisherApplication.java

package com.springboot;

//import statements

@SpringBootApplication
@EnableDiscoveryClient
@EnableJms
public class PublisherApplication {

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


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

    @Bean
    public RestTemplate restTemplate(RestTemplateBuilder builder) {
        return builder.build();

    }

}

发布消息.java
[发布主题的Restendpoint]

package com.springboot.controller;

//import statements

@RestController
@RequestMapping(path = "/schoolDashboard/topic")
class PublishMessage {

    public static final String MAILBOX_TOPIC = "mailbox.topic";

    @Autowired
    private JmsTemplate jmsTemplate;

    @GetMapping(path = "/sendEmail")
    public void sendStudentById() throws Exception{
        System.out.println("Anindya-TopicSendMessage.java :: Publishing Email sent....");
        jmsTemplate.convertAndSend(MAILBOX_TOPIC, "Topic - Email Sent");
    }

}

接收器应用程序01
[注意 - 接收器01是第一个微服务]

package com.springboot;

//import statements

@SpringBootApplication
@EnableJms
public class ReceiverApplication01 {


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

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

}

TopicMesssgeReceiver01.java
[Receiver01从主题读取消息]

package com.springboot.message;

//import statement

@Component
public class TopicMesssgeReceiver01 {

    private final SimpleMessageConverter converter = new SimpleMessageConverter();

    public static final String MAILBOX_TOPIC = "mailbox.topic";

    @JmsListener(destination = MAILBOX_TOPIC, containerFactory = "topicListenerFactory")
    public void receiveMessage(final Message message) throws JMSException{
        System.out.println("Receiver01 <" + String.valueOf(this.converter.fromMessage(message)) + ">");
    }

}

ReceiverApplication 02
[注意：-Receiver02是第二个微服务]

package com.springboot;

//import statement

@SpringBootApplication
@EnableJms
public class ReaderApplication02 {

    @Bean
    public JmsListenerContainerFactory<?> topicListenerFactory(ConnectionFactory connectionFactory,
            DefaultJmsListenerContainerFactoryConfigurer configurer) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setPubSubDomain(true);
        configurer.configure(factory, connectionFactory);       
        return factory;
    }

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

}

主题消息接收者02
[接收者02 从主题读取消息]

package com.springboot.message;

//import statement

@Component
public class TopicMesssgeReceiver02 {


private final SimpleMessageConverter converter = new SimpleMessageConverter();

    public static final String MAILBOX_TOPIC = "mailbox.topic";

    @JmsListener(destination = MAILBOX_TOPIC, containerFactory = "topicListenerFactory")
    public void receiveMessage(final Message message) throws Exception{
        System.out.println("Receiver02 <" + String.valueOf(this.converter.fromMessage(message)) + ">");
    }

}

Answer 1

谢谢你纳文。！最后，我能够做到这一点。< br >我们必须只设置setubsubdomain(true)；spring-boot会处理所有锅炉板代码。< br >现在，两个接收器微服务可以同时从主题中读取消息< br >以下是代码更改

发布消息.java
[发布主题的Restendpoint]

package com.springboot.controller;

//import statements

@RestController
@RequestMapping(path = "/schoolDashboard/topic")
class PublishMessage {

    public static final String MAILBOX_TOPIC = "mailbox.topic";

    @Autowired
    private JmsTemplate jmsTemplate;

    @GetMapping(path = "/sendEmail")
    public void sendStudentById() throws Exception{
        System.out.println("Publisher :: Message sent...");
        /* Added this statement. setPubSubDomain(true) identifies Topic in ActiveMQ */
        jmsTemplate.setPubSubDomain(true);
        jmsTemplate.convertAndSend(MAILBOX_TOPIC, "Topic - Email Sent");
    }

}

ReceiverApplication 02
[注意：-Receiver02是第二个微服务]

package com.springboot;

//import statement

@SpringBootApplication
@EnableJms
public class ReaderApplication02 {

    @Bean
    public JmsListenerContainerFactory<?> topicListenerFactory(ConnectionFactory connectionFactory,
            DefaultJmsListenerContainerFactoryConfigurer configurer) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();       
        configurer.configure(factory, connectionFactory);    
        /* setPubSubDomain(true) should be placed after 
         * configuration of the specified jms listener container factory*/
        factory.setPubSubDomain(true);
        return factory;
    }

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

}