spring boot jms - 在@JmsListner 中发送和接收消息

Question

QUEUE_INI 使用 IBM MQ9、最新的 spring boot 和 IBM 的 spring-boot-starter。

我想用@JmsListner 接收一条消息，然后不提交它，将另一条消息发送到另一个队列，接收响应，然后全部提交。

到目前为止，我还有下一个：

    @JmsListener(destination = QUEUE_IN, selector = "type='com.example.MainRequest")
    public void receiveMessage(Message message) throws JMSException {
        LOG.info("Received {} <{}>", QUEUE_IN, message);
        jmsTemplate2.convertAndSend(QUEUE_OUT, "SUB Request", requestMsg -> {
            requestMsg.setStringProperty("type", "com.example.SubRequest");
            return requestMsg;
        });
        LOG.info("Message sent");
        Message reply = jmsTemplate2.receiveSelected(QUEUE_IN, "type='com.example.SubResponse'");
        LOG.info("Received reply from {}: <{}>", QUEUE_IN, reply);
    }

我卡在“已发送消息”上。看起来子请求还没有真正发送。我在 MQ UI 中看到队列深度为 1，但里面没有消息，我的子请求侦听器也没有看到任何消息。

我也试过使用sendAndReceive 方法：

Message reply = jmsTemplate.sendAndReceive(QUEUE_OUT, session -> {
    Message msg = session.createTextMessage();
    msg.setStringProperty("type", "com.example.SubRequest");
    LOG.info("Sending msg: <{}> to {}", msg, QUEUE_OUT);
    return msg;
});

但我没有访问模型队列的权限。

有什么办法可以做到吗？

更新： 我在大家的共同帮助下完成了这项工作。我最终得到了单独的服务，只发送带有@Transactional(propagation = Propagation.REQUIRES_NEW) 的子请求。所有其他逻辑都保留在主侦听器中。

同时打开事务开始/结束日志也很有帮助：

logging:
  level:
    org.springframework.transaction.interceptor: trace

Answer 1

您的出站消息发送者与您的消息接收者在同一事务中注册。因此，出站消息的接收者在事务提交之前不会看到该消息。我认为您需要启动一个新事务来执行内部过程。

== 更新 ==

所以已经有一段时间了，我没有为此设置开发环境，但我建议这样。本质上，您将 Listener 和 Sender/Receiver 分成两个单独的类。 Sender/Receiver 应该被注入到你的监听器中，这样@Transactional 注释才会被尊重。

public class MyListener {
    private final MySender sender;

    public MyListener(MySender sender) {
        this.sender = sender;
    }

    @JmsListener(destination = QUEUE_IN, selector = "type='com.example.MainRequest")
    public void receiveMessage(Message message) throws JMSException {
        LOG.info("Received {} <{}>", QUEUE_IN, message);
        Message reply = sender.sendAndReceive()
        LOG.info("Received reply from {}: <{}>", QUEUE_IN, reply);
    }
}


public class MySender {
    private final JmsTemplate jmsTemplate2;
    private final Destination QUEUE_OUT;

    public MySender(JmsTemplate jmsTemplate2, Destination QUEUE_OUT) {
        this.jmsTemplate2 = jmsTemplate2;
        this.QUEUE_OUT = QUEUE_OUT;
    }

    @Transactional(propagation=Propagation.NESTED)  // Or REQUIRES_NEW, edepending on usage
    public Message sendAndReceive() throws JMSException {
        jmsTemplate2.convertAndSend(QUEUE_OUT, "SUB Request", requestMsg -> {
            requestMsg.setStringProperty("type", "com.example.SubRequest");
            return requestMsg;
        });
        LOG.info("Message sent");
        Message reply = jmsTemplate2.receiveSelected(QUEUE_IN, "type='com.example.SubResponse'");
        LOG.info("Received reply from {}: <{}>", QUEUE_IN, reply);
        return reply;
    }
}

@Transactional(propagation=Propagation.NESTED) 将启动一个新事务，如果一个已经在运行（并在方法退出时提交/回滚），因此您应该能够发送/接收消息并将其返回给您听众。

Answer 2

很遗憾，由于我没有足够的权限，我无法对上一个答案添加评论。

正如@Nicholas 所说，您的消息接收与消息发送在同一事务中。在receiveMessage 方法完成之前，初始发送不会被提交到队列以供接收者使用。我假设您可能将接收配置为 RECEIVE_TIMEOUT_INDEFINITE_WAIT 之类的东西，然后会阻止该方法的完成。

使用此配置，您已经有一个绑定到QUEUE_IN 的消息侦听器。传递到该队列的消息将仅传递给一个消费者。我看到你正在使用选择器来避免这个问题。

一个选项可能是为 type = com.example.SubResponse 引入另一个消息侦听器并从第一个中删除阻塞接收。