学步园

 

package test.requestreply;

 

import javax.jms.ConnectionFactory;

 

import org.apache.activemq.ActiveMQConnectionFactory;

 

import org.apache.activemq.camel.component.ActiveMQComponent;

 

import org.apache.camel.CamelContext;

 

import org.apache.camel.Exchange;

 

import org.apache.camel.Processor;

 

import org.apache.camel.ProducerTemplate;

 

import org.apache.camel.builder.RouteBuilder;

 

import org.apache.camel.component.mock.MockEndpoint;

 

import org.apache.camel.impl.DefaultCamelContext;

 

public class Test {

publicstaticvoid
main(String args[])throws Exception {

fromJava();

}

publicstaticvoid
fromJava()throws Exception {

CamelContext context = new DefaultCamelContext();

ProducerTemplate template = context.createProducerTemplate();

ActiveMQComponent component = context.getComponent("activemq", ActiveMQComponent.class);

ConnectionFactory cf = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false&broker.useJmx=false");

component.setConnectionFactory(cf);

context.addRoutes(new RouteBuilder() {

@Override

 

publicvoid
configure()throws Exception {

from("activemq:queue:queue2").inOut("activemq:queue:queue3").process(new
TestProcessor())

.to("mock:result").to("log:com.citi.etrading.retail.tps.batch.common.processor.ETPSBatchMessageConverter?level=INFO");

from("activemq:queue:queue3").process(new
TestProcessor1())

.to("mock:result");

　

from("activemq:queue:queue1?replyTo=queue:queue2&replyToType=Exclusive")

.transform(constant("Bye Camel"));

//.to("activemq:queue:queue3");

 

}

});

context.start();

template.sendBody("activemq:queue:queue1","Hello
World");

Thread.sleep(5000);

MockEndpoint endpoint = (MockEndpoint)context.getEndpoint("mock:result");

endpoint.expectedBodiesReceived("Hello World");

context.stop();

}

publicstaticclass
TestProcessorimplements Processor{

publicvoid
process(Exchange exchange)throws Exception {

//TODO
Auto-generated method stub

 

System.out.println(exchange);

System.out.println("queue2");

}

}

publicstaticclass
TestProcessor1implements Processor{

publicvoid
process(Exchange exchange)throws Exception {

//TODO
Auto-generated method stub

 

System.out.println(exchange);

System.out.println("queue3");

}

}

}

　

 

 

===============

http://stackoverflow.com/questions/10085340/camel-is-it-possible-to-implement-request-reply-competing-consumers-with-the

If you use Camel 2.9 or better, then I suggest to use replyToType=Exclusive on the activemq endpoint where you do request/reply. This tells Camel that the queue is exclusive, and it speedup, as no JMS message selectors is needed to pickup expected correlated
messages.

See the section Request-reply over JMS onwards at the Camel JMS docs:http://camel.apache.org/jms

If you use temporary queues, then that is also fast as well, as no JMS message selectors is needed.

Also your route starts with a direct endpoint. That is a synchronous call, so the caller will wait/block until the Exchange is completely done.

Also the Splitter EIP can be configured to run in parallel mode which will use concurrent processing. And if you have a big message to split, then consider using streaming which will split the message on-demand, instead of loading the entire message content
into memory.

Anyway there is a lot going on in the route. Can you pin-point more precisely where you have an issue? It makes it easier to help out.

 

 

=================

http://grokbase.com/t/camel/users/128n88xeva/how-to-use-request-reply-in-jms

Hi

Just configure the name of the reply queue on the from uri. You may
want to configure the reply queue as being exclusive then it runs
faster. Then when the route completes, the message will be send back
to the reply queue. eg in the example below, after the processor has
run, the message containing "Hello World" will be send back to the
OUTPUT.Q.

from("activemq://queue:INPUT.Q?jmsMessageType=Object&replyTo=OUTPUT.Q&replyToType=Exclusive")
.process(new
Processor() {
@Override
public void process(Exchange exchange) throws Exception {
exchange.getOut().setBody("Hello World");
}
});

I suggest to read the JMS page. It has sections about request/reply
etc, and some pointers how and what to look out for. And what to
configure etc.
http://camel.apache.org/jms

And mind that you can use 3 kind of queues when doing request/reply
- temporary queue
- shared queue
- exclusive queue

So mind the difference.

Make sure someone is listening on OUTPUT.Q and sending back a message
to REPLY.Q.
You get this exception because Camel timed out after 20 seconds. You
can adjust this value if you need longer timeout.

===================

http://camel.465427.n5.nabble.com/After-publish-OutOnly-Reply-received-for-unknown-correlationID-td3368679.html

After publish (OutOnly): Reply received for unknown correlationID

Use the disableReplyTo option to not expect any replies.

disableReplyTo false If true, a producer will behave like a InOnly exchange with the exception that JMSReplyTo header is sent out and not be suppressed like in the case of InOnly. Like InOnly the producer will not wait for a reply. A consumer with this flag
will behave like InOnly. This feature can be used to bridge InOut requests to another queue so that a route on the other queue will send it´s response directly back to the original JMSReplyTo.

 

================http://camel.apache.org/jms.html

About using Camel to send and receive messages and JMSReplyTo

The JMS component is complex and you have to pay close attention to how it works in some cases. So this is a short summary of some of the areas/pitfalls to look for.

When Camel sends a message using its JMSProducer, it checks the following conditions:

• The message exchange pattern,
• Whether a JMSReplyTo was set in the endpoint or in the message headers,
• Whether any of the following options have been set on the JMS endpoint:
disableReplyTo, preserveMessageQos, explicitQosEnabled.

All this can be a tad complex to understand and configure to support your use case.

JmsProducer

The JmsProducer behaves as follows, depending on configuration:

JmsConsumer

The JmsConsumer behaves as follows, depending on configuration:

So pay attention to the message exchange pattern set on your exchanges.

If you send a message to a JMS destination in the middle of your route you can specify the exchange pattern to use, see more at

Request Reply.

This is useful if you want to send an InOnly message to a JMS topic: