如何从ActiveMQ队列创建Spring反应器通量？

Question

我正在使用SpringReactor3组件和SpringIntegration进行实验，以从JMS队列创建反应流（Flux）。

我试图从JMS队列（使用SpringIntegration的ActiveMQ）创建一个反应流（SpringReactor3Flux），以供客户端异步获取JMS消息。我相信我已经正确连接了所有内容，但在服务器停止之前，客户端不会接收任何JMS消息。然后，所有消息都被“推送”到客户端a一次。

任何帮助都将不胜感激。

这是我用来配置JMS、集成组件和反应式发布者的配置文件：

@Configuration
@EnableJms
@EnableIntegration
public class JmsConfiguration {

    @Value("${spring.activemq.broker-url:tcp://localhost:61616}")
    private String defaultBrokerUrl;

    @Value("${queues.patient:patient}")
    private String patientQueue;

    @Autowired
    MessageListenerAdapter messageListenerAdapter;

    @Bean
    public DefaultJmsListenerContainerFactory myFactory(
            DefaultJmsListenerContainerFactoryConfigurer configurer) {
        DefaultJmsListenerContainerFactory factory =
                new DefaultJmsListenerContainerFactory();
        configurer.configure(factory, jmsConnectionFactory());
        return factory;
    }

    @Bean
    public Queue patientQueue() {
        return new ActiveMQQueue(patientQueue);

    }

    @Bean
    public ActiveMQConnectionFactory jmsConnectionFactory() {
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        connectionFactory.setBrokerURL(defaultBrokerUrl);
        connectionFactory.setTrustedPackages(Arrays.asList("com.sapinero"));
        return connectionFactory;
    }

    // Set the jackson message converter
    @Bean
    public JmsTemplate jmsTemplate() {
        JmsTemplate template = new JmsTemplate();
        template.setConnectionFactory(jmsConnectionFactory());
        template.setDefaultDestinationName(patientQueue);
        template.setMessageConverter(jacksonJmsMessageConverter());
        return template;
    }

    @Bean
    public MessageListenerAdapter messageListenerAdapter() {
        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter();
        messageListenerAdapter.setMessageConverter(jacksonJmsMessageConverter());
        return messageListenerAdapter;
    }

    @Bean
    public AbstractMessageListenerContainer messageListenerContainer() {
        DefaultMessageListenerContainer defaultMessageListenerContainer = new DefaultMessageListenerContainer();
        defaultMessageListenerContainer.setMessageConverter(jacksonJmsMessageConverter());
        defaultMessageListenerContainer.setConnectionFactory(jmsConnectionFactory());
        defaultMessageListenerContainer.setDestinationName(patientQueue);
        defaultMessageListenerContainer.setMessageListener(messageListenerAdapter());
        defaultMessageListenerContainer.setCacheLevel(100);
        defaultMessageListenerContainer.setErrorHandler(new ErrorHandler() {
            @Override
            public void handleError(Throwable t) {
                t.printStackTrace();
            }
        });

        return defaultMessageListenerContainer;
    }

    @Bean // Serialize message content to json using TextMessage
    public MessageConverter jacksonJmsMessageConverter() {
        MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
        converter.setTargetType(MessageType.TEXT);
        converter.setTypeIdPropertyName("_type");
        return converter;
    }


    @Bean
    public MessageChannel jmsOutboundInboundReplyChannel() {
        return MessageChannels.queue().get();
    }

    @Bean
    public Publisher<Message<String>> pollableReactiveFlow() {
        return IntegrationFlows
                .from(Jms.messageDrivenChannelAdapter(messageListenerContainer()).get())
                .channel(MessageChannels.queue())
                .log(LoggingHandler.Level.DEBUG)
                .log()
                .toReactivePublisher();
    }

    @Bean
    public MessageChannel jmsChannel() {
        return new DirectChannel();
    }

创建Flux的控制器是：

@RestController
@RequestMapping("patients")
public class PatientChangePushController {
    private LocalDateTime lastTimePatientDataRetrieved = LocalDateTime.now();
    private int durationInSeconds = 30;
    private Patient patient;
    AtomicReference<SignalType> checkFinally = new AtomicReference<>();

    @Autowired
    PatientService patientService;

    @Autowired
    @Qualifier("pollableReactiveFlow")
    private
    Publisher<Message<String>> pollableReactiveFlow;

    @Autowired
    private JmsTemplate jmsTemplate;

    @Autowired
    private Queue patientQueue;

    /**
     * Subscribe to a Flux of a patient that has been updated.
     *
     * @param id
     * @return
     */
    @GetMapping(value = "/{id}/alerts", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<Message<String>> getPatientAlerts(@PathVariable Long id) {

        Flux<Message<String>> messageFlux = Flux.from(pollableReactiveFlow);
        return messageFlux;
    }

    @GetMapping(value = "/generate")
    public void generateJmsMessage() {
        for (long i = 0L; i < 100; i++) {
            Patient patient = new Patient();
            patient.setId(i);
            send(patient);
            System.out.println("Message was sent to the Queue");
        }

    }

    void send(Patient patient) {
        this.jmsTemplate.convertAndSend(this.patientQueue, patient);
    }

}

如果有人能告诉我为什么消息直到服务器被杀死后才发送到客户端，我将不胜感激。

Answer 1

对我很管用：

@SpringBootApplication
@RestController
public class SpringIntegrationSseDemoApplication {

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

    @Autowired
    private ConnectionFactory connectionFactory;

    @Autowired
    private JmsTemplate jmsTemplate;

    @Bean
    public Publisher<Message<String>> jmsReactiveSource() {
        return IntegrationFlows
                .from(Jms.messageDrivenChannelAdapter(this.connectionFactory)
                        .destination("testQueue"))
                .channel(MessageChannels.queue())
                .log(LoggingHandler.Level.DEBUG)
                .log()
                .toReactivePublisher();
    }

    @GetMapping(value = "/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> getPatientAlerts() {
        return Flux.from(jmsReactiveSource())
                .map(Message::getPayload);
    }

    @GetMapping(value = "/generate")
    public void generateJmsMessage() {
        for (int i = 0; i < 100; i++) {
            this.jmsTemplate.convertAndSend("testQueue", "testMessage #" + (i + 1));
        }
    }

}

在一个终端中，我有< code > curl http://localhost:8080/events ，它等待来自< code>Flux的SSEs。

在其他终端中，我执行卷曲 http://localhost:8080/generate，并在第一个终端中看到：

data:testMessage #1

data:testMessage #2

data:testMessage #3

data:testMessage #4

我使用Spring Boot 2.0.0.BUILD-SNAPSHOT。

另请参见此处：https://spring.io/blog/2017/03/08/spring-tips-server-sent-events-sse