从java中的骆驼路线接收消息

Question

为了使用我部署的包含骆驼路由的捆绑包作为我的中间件，我想将消息发送到骆驼路由，然后发送到 cxf 端点。记录响应。现在，如果我使用MessageConsumer，我的外部应用程序将无法收到来自骆驼路线的响应。

有没有办法在我的主程序中从骆驼路线获取响应消息并打印出来？

Answer 1

以下是我的最终路线。它接收来自外部应用程序的请求，向 cxf 端点发送 webservice 请求，接收并将响应发送回 inonly 队列，在外部应用程序中使用。

from("activemq:queue:fork-customers")
                .routeId("activemq:queue:fork-customers")
                .setExchangePattern(ExchangePattern.InOut)
                .convertBodyTo(String.class)
                .process(new Processor() {
                    public void process(Exchange exchange) throws Exception {
                        Document doc = DocumentBuilderFactory.newInstance().newDocumentBuilder()
                                .parse(new InputSource(new StringReader((String) exchange.getIn().getBody())));
                        exchange.getIn().setBody(doc);
                    }
                })
                .to("freemarker:Envelope.ftl")
                .setHeader("operationName", simple("findCustomer"))
                .to("cxf:bean:my-webservice?dataFormat=PAYLOAD")
                .to("log:reply")
                .process(new Processor() {
                    public void process(Exchange exchange) throws Exception {
                        Logger log = LoggerFactory.getLogger(XmlRouting.class);
                        Message msg = exchange.getIn();
                        log.info("CXF Response : " +msg.toString());                        
                    }
                })
                .to("file://E://Target//Response")
                .inOnly("activemq:queue:jmsResponse");

产生消息并发送到activemq并通过inonly activemq接收响应的外部应用程序代码。

ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                "tcp://localhost:61616");
        // Create a Connection
        String userName = "smx";
        String password = "smx";
        Connection connection = connectionFactory.createConnection(userName,
                password);
        connection.start();
        // Create a Session
        Session session = connection.createSession(false,
                Session.AUTO_ACKNOWLEDGE);
        // Create the destination (Queue)
        Queue destination = session.createQueue("fork-customers");
        // Create a MessageProducer from the Session to the Topic or Queue
        MessageProducer producer = session.createProducer(destination);
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        // pass the arguements here
        TextMessage textMessage = session
                .createTextMessage("<root><arg0>CUST1001</arg0></root>");
        // Tell the producer to send the message
        Queue tempQueue = session.createQueue("jmsResponse");
        textMessage.setJMSReplyTo(tempQueue);
        producer.send(textMessage);
        MessageConsumer consumer = session.createConsumer(tempQueue);
        Message response = consumer.receive();
        String text;
        if (response instanceof TextMessage) {
            text = ((TextMessage) response).getText();
        } else {
            byte[] body = new byte[(int) ((BytesMessage) response)
                    .getBodyLength()];
            ((BytesMessage) response).readBytes(body);
            text = new String(body);
        }
        System.out.println("responseMsg " + text);
        // Clean up
        session.close();
        connection.close();