我正在尝试使用Spring Reactor 3组件和Spring Integration创建一个反应流(Flux)，从JMS队列中获取数据。 我试图为客户端异步获取JMS消息，创建一个反应流(Spring Reactor 3 Flux)从JMS队列(使用Spring Integration的ActiveMQ)。 我相信我已经正确地连接了所有组件，但是客户端在服务器停止之前不会接收到任何JMS消息。然后所有消息一次性“推送”到客户端。 如有帮助，请提供帮助。 这是我用来配置JMS、Integration组件和反应式发布者的配置文件:
创建Flux的控制器是：
如果有人能告诉我为什么消息直到服务器关闭后才发送给客户端，我将不胜感激。
@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);
}
}
如果有人能告诉我为什么消息直到服务器关闭后才发送给客户端，我将不胜感激。