【问题标题】:Update: Spring Boot JMS static reply queue on IBM MQ Series更新:IBM MQ 系列上的 Spring Boot JMS 静态回复队列
【发布时间】: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


【解决方案1】:

默认的 JmsTemplate(由JmsMessagingTemplate 使用)始终使用临时回复队列。您可以将其子类化并覆盖 doSendAndReceive(Session session, Destination destination, MessageCreator messageCreator) 以改用您的托管队列。

但是,它仅在您一次有一个未完成的请求时才有效(例如,所有请求都在单个线程上运行)。您还必须添加代码以通过检查相关 ID 来丢弃“迟到”的到达。

您可以改用异步发送并在侦听器容器上处理回复并将回复与请求相关联。

考虑改用spring-integration-jms 及其出站网关 - 它在回复队列处理方面具有更大的灵活性(并为您完成所有相关工作)。

https://docs.spring.io/spring-integration/reference/html/jms.html#jms-outbound-gateway

【讨论】:

    【解决方案2】:

    您缺少队列管理器。

    ibm:
      mq:
        queueManager: QM1
        channel: chanel
        connName: localhost(1414)
        user: admin
        password: admin
    

    【讨论】:

    • 队列管理器是channel属性的一部分,我写在后面的评论里。
    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2022-12-01
    • 1970-01-01
    • 2020-09-29
    • 2015-11-24
    • 2022-01-07
    • 1970-01-01
    相关资源
    最近更新 更多