0

我正在测试将 JMS 请求/回复与 Camel 和 ActiveMQ 一起使用的示例。当骆驼为您创建侦听器时，我可以使示例正常工作。IE。

from("direct:entryPoint").inOut("jms:queue:A");

from("jms:queue:A").
    process(new Processor() {

        @Override
        public void process(Exchange exchange) throws Exception {
            exchange.getIn().setBody("Hello World.");

        }
    });



我现在遇到的问题是我无法让 JMS 请求/回复与存在于 Camel 的 jvm 之外的 MessageListener 一起工作。连接超时等待回复。我确保 MessageListener 正在将回复发送到 replyTo 队列，并且我还在设置相关 ID。我在这里做错了什么？我已经用谷歌搜索了几天试图解决这个问题，但没有运气。提前感谢您的帮助。

下面是我正在使用的路线，我还将 MessageListener 逻辑也放在下面。

from("direct:entryPoint").
  inOut("jms:queue:B?concurrentConsumers=4&requestTimeout=240000");

队列 B 的 MessageListener onMessage：

@Override
public void onMessage(Message message) {

    String msg = null;
    ObjectMapper mapper = new ObjectMapper();
    mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, true);
    String jsonOutput = null;

    try{
        msg = ((TextMessage) message).getText();

        //convert message payload to purchase order
        PurchaseOrder order = mapper.readValue(msg, PurchaseOrder.class);

        //Set the id to see if the request reply worked.
        order.setOrderId(BigInteger.valueOf(111111111));


        if(message.getJMSReplyTo() != null){
            Map<String, Object> headers = new HashMap<String, Object>();
            headers.put("JMSCorrelationID", message.getJMSCorrelationID());
            headers.put("JMSReplyTo", message.getJMSReplyTo().toString());  
            jsonOutput = mapper.writeValueAsString(order);

            //Camel runs on the external jvm so leverage the producerTemplate.
            producerTemplate.
              sendBodyAndHeaders("jms:"+ message.getJMSReplyTo().toString(), 
              jsonOutput, headers);
        }
    }
    catch(Exception e){
        logger.fatal(e.getMessage());

        try {
            if(message.getJMSReplyTo() != null){
                Map<String, Object> headers = new HashMap<String, Object>();
                headers.put("JMSCorrelationID", message.getJMSCorrelationID());

                producerTemplate.
                 sendBodyAndHeaders("jms:"+ message.getJMSReplyTo().toString(),
 e.getMessage(), headers);
            }
        } catch (JMSException e1) {
            logger.fatal(e1.getMessage());
        } catch (Exception e1) {
            logger.fatal(e1.getMessage());
        }
    }
}
4

1 回答 1

1 
sendBodyAndHeaders("jms:"+ message.getJMSReplyTo().toString(), 
          jsonOutput, headers);

通常会解决jms:queue://someQueue可能会搞砸的问题。除非您采取预防措施，否则使用转换为字符串的 javax.jms.Destination 通常不是一个好主意。

您可以使用 Camel 标头CamelJmsDestination： headers.put("CamelJmsDestination",message.getJMSReplyTo());

这是否有效，我不知道。通常，尝试将 ActiveMQ 与 Web 控制台一起使用（或使用 JMX (jconsole) 连接到 ActiveMQ）来查看队列并尝试找出谁在读取哪个队列以及消息在哪里结束。这真的很有帮助。

于 2013-03-23T18:23:44.110 回答