Mailing-List: contact users-help@camel.apache.org; run by ezmlm
Precedence: bulk
Reply-To: users@camel.apache.org
Received-SPF: pass (nike.apache.org: domain of claus.ibsen@gmail.com
 designates 209.85.161.173 as permitted sender)
MIME-Version: 1.0
In-Reply-To: 
 <CAGO7Ob0YgFGVLr=TjEm-5KiHsxACCgQQW5YONak6VpDhUzrkpA@mail.gmail.com>
References: <4E165B00.200@referentia.com>
 <CAGB5yNmwemea-RsazM9Rg2REZM=8Q9pHqQieAnYThgdgfZwtmQ@mail.gmail.com>
 <CAGO7Ob0YgFGVLr=TjEm-5KiHsxACCgQQW5YONak6VpDhUzrkpA@mail.gmail.com>
From: Claus Ibsen <claus.ibsen@gmail.com>
Date: Sun, 14 Aug 2011 09:10:39 +0200
Message-ID: 
 <CAGB5yNnUMOq66VTvxbqfE7ZEWQTKUuM9eqt1h1eL=_rXYKM02Q@mail.gmail.com>
Subject: Re: slow reply for jms component when url contains replyTo
To: users@camel.apache.org
Content-Type: text/plain; charset=ISO-8859-1
Content-Transfer-Encoding: quoted-printable

On Sat, Aug 13, 2011 at 8:06 PM, David Karlsen <davidkarlsen@gmail.com> wro=
te:
> Hi - I would be very interested in this (and willing to test).
> We had to back out from using camel for our req/res jms (we now use
> jmstemplate directly) - especially because the persistentrelyqueuemanager
> (or something like that) seemed to use the same connection - so eventuall=
y
> the fw closed that connection during hours with little traffic.

The reply manager uses the default message listener which supports
automatic reconnection.
That was not the case in older Camel releases, so you may have hit
that issue back then.


>
> Another positive effect of switching to using jmstemplate directly is tha=
t
> the same thread is handling the response and the request - and we have a
> thread scoped state object attached to it.
>

You can store information on the Exchange then its kept safe for async
messaging / thread context switches etc.


> We are running in a cluster - but because we use correlationid this is no=
t a
> problem as you say.
>

So if you use a jmstemplate to do request/reply then you are not as
efficient as you will have threads blocking.
And as well having multiple threads racing for replies. So if you have
like 5 concurrent request/replies going on, then
you have 5 threads using the jmstemplate, and all 5 is also active
listening for the replies on the queue.
And you then use a shared correlation map to ensure each of those 5
gets the intended reply.

Is that correct?


> The underlying MOM is WebSphere MQ.
>
> BTW: "vinning" in my rely was supposted to spell "common" - I blame my ph=
one
> for that ;-)
>
> 2011/8/13 Claus Ibsen <claus.ibsen@gmail.com>
>
>> Hi Jim, others
>>
>> I got some code for supporting exclusive replyTo queues. I just want
>> to know if you run in a clustered environment.
>> As the replyTo queue would be exclusive to each CamelContext (eg each
>> node). So if you have 2 nodes running with Camel then each node would
>> have to use an unique replyTo queue.
>>
>> If you run in a single instance of CamelContext then there is of
>> course no problem.
>>
>> Alternative you can always use shared queues in a cluster, as each
>> CamelContext (each node) will only pickup intended reply message for
>> it (using a JMS selector), which is also the case why its slower. You
>> can tweak the performance with the receiveTimeout. That is default 1
>> sec. If you set it lower, then it will potential react faster to new
>> replies coming back.
>>
>>
>>
>> On Fri, Jul 8, 2011 at 3:18 AM, Jim Newsham <jnewsham@referentia.com>
>> wrote:
>> >
>> > I'm using Camel 2.7.1 on top of ActiveMQ 5.5.0. =A0For some reason, wh=
en I
>> > specify a custom replyTo destination on the endpoint url, the time it
>> takes
>> > for the producer to receive a reply increases drastically. =A0The curi=
ous
>> > thing is that the time to receive a reply is almost exactly 1 second.
>> =A0When
>> > I remove the replyTo from the url, everything's fast again.
>> >
>> > I created a very simple, stand-alone test to demonstrate what I'm seei=
ng.
>> > =A0There is a server class [4] which runs an embedded instance of Acti=
veMQ
>> and
>> > simply replies to messages as they arrive; and a client [3] class whic=
h
>> > simply sends messages to the server, and prints the elapsed time. =A0T=
he
>> > USE_REPLY_TO symbolic constant in the client determines whether a repl=
yTo
>> > value is added to the url or not.
>> >
>> > The client output when USE_REPLY_TO is false is shown as [1]. =A0The c=
lient
>> > output when USE_REPLY_TO is true is shown as [2]. =A0The code is prett=
y
>> > trivial. =A0Am I doing something wrong, or is this a Camel and/or Acti=
veMQ
>> > issue?
>> >
>> > Thanks!
>> > Jim
>> >
>> >
>> > [1] USE_REPLY_TO =3D false
>> >
>> > received reply in: 0.476 s
>> > received reply in: 0.006 s
>> > received reply in: 0.006 s
>> > received reply in: 0.006 s
>> > received reply in: 0.006 s
>> > ...
>> >
>> >
>> > [2] USE_REPLY_TO =3D true
>> >
>> > received reply in: 1.524 s
>> > received reply in: 1.002 s
>> > received reply in: 1.003 s
>> > received reply in: 1.003 s
>> > received reply in: 1.002 s
>> > ...
>> >
>> >
>> > [3] TestReplyToClient.java
>> >
>> > package test;
>> >
>> > import org.apache.activemq.ActiveMQConnectionFactory;
>> > import org.apache.activemq.camel.component.ActiveMQComponent;
>> > import org.apache.camel.CamelContext;
>> > import org.apache.camel.ProducerTemplate;
>> > import org.apache.camel.impl.DefaultCamelContext;
>> >
>> > public class TestReplyToClient {
>> >
>> > =A0private static final boolean USE_REPLY_TO =3D false;
>> >
>> > =A0public static void main(String... args) throws Exception {
>> > =A0 =A0// create camel context; configure activemq component for
>> > tcp://localhost:7001
>> > =A0 =A0CamelContext context =3D new DefaultCamelContext();
>> > =A0 =A0ActiveMQComponent activemqComponent =3D
>> > ActiveMQComponent.activeMQComponent();
>> > =A0 =A0activemqComponent.setConnectionFactory(new ActiveMQConnectionFa=
ctory(
>> > =A0 =A0 =A0null, null, "tcp://localhost:7001"));
>> > =A0 =A0context.addComponent("activemq", activemqComponent);
>> > =A0 =A0context.start();
>> >
>> > =A0 =A0// define url to send requests to
>> > =A0 =A0String sendUrl =3D "activemq:queue:dest";
>> > =A0 =A0if (USE_REPLY_TO) {
>> > =A0 =A0 =A0sendUrl +=3D "?replyTo=3DreplyQueue";
>> > =A0 =A0}
>> > =A0 =A0System.err.println("sending to url: " + sendUrl);
>> >
>> > =A0 =A0// repeatedly send requests; measure elapsed time
>> > =A0 =A0ProducerTemplate template =3D context.createProducerTemplate();
>> > =A0 =A0while (true) {
>> > =A0 =A0 =A0long startNanos =3D System.nanoTime();
>> > =A0 =A0 =A0template.requestBody(sendUrl, "abc");
>> > =A0 =A0 =A0long elapsedNanos =3D System.nanoTime() - startNanos;
>> > =A0 =A0 =A0System.err.println(String.format("received reply in: %.3f s=
",
>> > elapsedNanos / 1000000000.0));
>> > =A0 =A0}
>> > =A0}
>> >
>> > }
>> >
>> >
>> > [4] TestReplyToServer.java
>> >
>> > package test;
>> >
>> > import org.apache.activemq.broker.BrokerService;
>> > 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.builder.RouteBuilder;
>> > import org.apache.camel.impl.DefaultCamelContext;
>> >
>> > public class TestReplyToServer {
>> >
>> > =A0private static final String BROKER_NAME =3D "thebroker";
>> >
>> > =A0public static void main(String... args) throws Exception {
>> > =A0 =A0startBroker();
>> > =A0 =A0startCamel();
>> > =A0 =A0Thread.sleep(Long.MAX_VALUE);
>> > =A0}
>> >
>> > =A0private static void startBroker() throws Exception {
>> > =A0 =A0BrokerService brokerService =3D new BrokerService();
>> > =A0 =A0brokerService.setBrokerName(BROKER_NAME);
>> > =A0 =A0brokerService.setSchedulerSupport(false);
>> > =A0 =A0brokerService.setPersistent(false);
>> > =A0 =A0brokerService.addConnector("tcp://0.0.0.0:7001");
>> > =A0 =A0brokerService.start();
>> > =A0 =A0brokerService.waitUntilStarted();
>> > =A0}
>> >
>> >
>> > =A0private static void startCamel() throws Exception {
>> > =A0 =A0CamelContext context =3D new DefaultCamelContext();
>> >
>> > =A0 =A0ActiveMQComponent activemqComponent =3D
>> > ActiveMQComponent.activeMQComponent();
>> > =A0 =A0activemqComponent.setBrokerURL(String.format("vm://%s?create=3D=
false",
>> > BROKER_NAME));
>> > =A0 =A0context.addComponent("activemq", activemqComponent);
>> >
>> > =A0 =A0final String receiveUrl =3D "activemq:queue:dest";
>> > =A0 =A0context.addRoutes(new RouteBuilder() {
>> > =A0 =A0 =A0@Override
>> > =A0 =A0 =A0public void configure() throws Exception {
>> > =A0 =A0 =A0 =A0from(receiveUrl).process(new Processor() {
>> > =A0 =A0 =A0 =A0 =A0@Override
>> > =A0 =A0 =A0 =A0 =A0public void process(Exchange exchange) throws Excep=
tion {
>> > =A0 =A0 =A0 =A0 =A0 =A0System.err.println("received request");
>> > =A0 =A0 =A0 =A0 =A0 =A0exchange.getOut().setBody("reply");
>> > =A0 =A0 =A0 =A0 =A0}
>> > =A0 =A0 =A0 =A0});
>> > =A0 =A0 =A0}
>> > =A0 =A0});
>> >
>> > =A0 =A0context.start();
>> > =A0 =A0System.err.println("listening on url: " + receiveUrl);
>> > =A0}
>> >
>> > }
>> >
>> >
>> >
>> >
>> >
>>
>>
>>
>> --
>> Claus Ibsen
>> -----------------
>> FuseSource
>> Email: cibsen@fusesource.com
>> Web: http://fusesource.com
>> Twitter: davsclaus, fusenews
>> Blog: http://davsclaus.blogspot.com/
>> Author of Camel in Action: http://www.manning.com/ibsen/
>>
>
>
>
> --
> --
> David J. M. Karlsen - http://www.linkedin.com/in/davidkarlsen
>


--=20
Claus Ibsen
-----------------
FuseSource
Email: cibsen@fusesource.com
Web: http://fusesource.com
Twitter: davsclaus, fusenews
Blog: http://davsclaus.blogspot.com/
Author of Camel in Action: http://www.manning.com/ibsen/