【发布时间】:2021-06-26 11:15:15
【问题描述】:
在我的用例中,我需要通过托管队列对远程系统进行请求-回复调用。使用 Spring Boot 和 IBM 的 MQ 启动器时,我遇到的问题是应用程序想要创建动态/临时回复队列,而不是使用已经存在的托管队列。
配置在这里设置
@EnableJms
@Configuration
public class QueueConfiguration {
@Bean
public MQQueueConnectionFactory connectionFactory() throws JMSException {
MQQueueConnectionFactory factory = new MQQueueConnectionFactory();
factory.setTransportType(CT_WMQ); // is 1
factory.setHostName(queueProperties.getHost());
factory.setPort(queueProperties.getPort());
factory.setChannel(queueProperties.getChannel()); // combo of ${queueManager}%${channel}
return factory;
}
@Bean
public JmsMessagingTemplate messagingTemplate(ConnectionFactory connectionFactory) {
JmsMessagingTemplate jmt = new JmsMessagingTemplate(connectionFactory);
jmt.setDefaultDestinationName(queueProperties.getQueueName());
return jmt;
}
@Bean
public Jaxb2Marshaller marshaller() {
Jaxb2Marshaller marshaller = new Jaxb2Marshaller();
marshaller.setPackagesToScan("com.foo.model");
return marshaller;
}
@Bean
public MessageConverter messageConverter(Jaxb2Marshaller marshaller) {
MarshallingMessageConverter converter = new MarshallingMessageConverter();
converter.setMarshaller(marshaller);
converter.setUnmarshaller(marshaller);
return converter;
}
}
用法非常简单:将对象转换并发送。等待响应接收 并转换它。
@Component
public class ExampleSenderReceiver {
@Autowired
private JmsMessagingTemplate jmsMessagingTemplate;
@Override
@SneakyThrows
public ResponseExample sendAndReceive(RequestExample request, String correlationId) {
MessagePostProcessor mpp = message -> {
message = MessageBuilder.fromMessage(message)
.setHeader(JmsHeaders.CORRELATION_ID, correlationId)
// .setHeader(JmsHeaders.REPLY_TO, "DEV.QUEUE.3") this triggers queue creation
.build();
return message;
};
String destination = Objects.requireNonNull(jmsMessagingTemplate.getDefaultDestinationName());
return jmsMessagingTemplate.convertSendAndReceive(destination, request, ResponseExample.class, mpp);
}
我已经阅读了大量 IBM 文档并认为,我需要将 message type 设置为“MQMT_REQUEST”,但我没有找到合适的位置。
更新
将 Spring Integration 添加为 Gary proposed 并添加了 JmsOutboundGateway 的配置
@Bean
public MessageChannel requestChannel() {
return new DirectChannel();
}
@Bean
public QueueChannel responseChannel() {
return new QueueChannel();
}
@Bean
@ServiceActivator(inputChannel = "requestChannel" )
public JmsOutboundGateway jmsOutboundGateway( ConnectionFactory connectionFactory) {
JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setConnectionFactory(connectionFactory);
gateway.setRequestDestinationName("REQUEST");
gateway.setReplyDestinationName("RESPONSE");
gateway.setReplyChannel(responseChannel());
gateway.setCorrelationKey("JMSCorrelationID*");
gateway.setIdleReplyContainerTimeout(2, TimeUnit.SECONDS);
return gateway;
}
并改编了我的 ExampleSenderReceiver 类
@Autowired
@Qualifier("requestChannel")
private MessageChannel requestChannel;
@Autowired
@Qualifier("responseChannel")
private QueueChannel responseChannel;
@Override
@SneakyThrows
public ResponseExample sendAndReceive(RequestExample request, String correlationId) {
String xmlContent = "the marshalled request object";
Map<String, Object> header = new HashMap<>();
header.put(JmsHeaders.CORRELATION_ID, correlationId);
GenericMessage<String> message1 = new GenericMessage<>(xmlContent, header);
requestChannel.send(message1);
log.info("send done" );
Message<?> receive = responseChannel.receive(1500);
if(null != receive){
log.info("incoming: {}", receive.toString());
}
}
重要的部分是gateway.setCorrelationKey("JMSCorrelationID*");
没有该行,correlationId 传播不正确。
下一步是重新添加 MessageConverters 并让它再次变得更好。
谢谢。
【问题讨论】:
-
另一种方法是在 jmsTemplate 上使用
send而不是sendAndReceive。记住在传出消息上设置 REPLY_TO。然后,您可以在对队列的回复中设置一个常规的listener。 -
这是另一个最初的想法,但随后需要缓存调用的上下文,并且必须手动关联响应。还需要仅收集属于请求的响应,因此来自网关的动态消息选择器非常适合此用例。
-
好点。我猜你可能需要使用消息头。请求添加消息头设置,响应者将其添加到回复中,侦听器在头设置上使用选择器。
标签: java spring-boot spring-integration ibm-mq spring-jms