I'm attaching a simplified test case in case anybody wants to have a
look (code attached at bottom). For the following test case, 12
requests are made. In the client, we log when a request is sent, and
when the reply is received. Each request completes instantaneously,
except for request 0 which requires 12 seconds. On the client, we
expect each reply to be returned immediately for each request, except
for request 0 which should be delayed 12 seconds.
Expected client output:
sending request: 0
sending request: 1
received reply for request: 1
sending request: 2
received reply for request: 2
sending request: 3
received reply for request: 3
sending request: 4
received reply for request: 4
sending request: 5
received reply for request: 5
sending request: 6
received reply for request: 6
sending request: 7
received reply for request: 7
sending request: 8
received reply for request: 8
sending request: 9
received reply for request: 9
sending request: 10
received reply for request: 10
sending request: 11
received reply for request: 11
sending request: 12
received reply for request: 0
received reply for request: 12
Actual client output:
sending request: 0
sending request: 1
received reply for request: 1
sending request: 2
received reply for request: 2
sending request: 3
sending request: 4
received reply for request: 4
sending request: 5
received reply for request: 5
sending request: 6
sending request: 7
received reply for request: 7
sending request: 8
received reply for request: 8
sending request: 9
sending request: 10
received reply for request: 10
sending request: 11
received reply for request: 11
sending request: 12
received reply for request: 0
received reply for request: 3
received reply for request: 6
received reply for request: 9
received reply for request: 12
Notice how request numbers 3, 6, 9 are delayed until request 0 has
completed processing. Since we have three threads, this suggests that
requests are being dispatched to threads in a round-robin fashion,
instead of having the threads compete to consume the requests from a
single queue.
Thanks,
Jim
Server and client code follow:
// SERVER
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 TestConcConsumersServer {
public static void main(String... args) throws Exception {
// start embedded activemq
String brokerName = "thebroker";
BrokerService brokerService = new BrokerService();
brokerService.setBrokerName(brokerName);
brokerService.setSchedulerSupport(false);
brokerService.setPersistent(false);
brokerService.addConnector("tcp://0.0.0.0:7001");
brokerService.start();
brokerService.waitUntilStarted();
// start camel with activemq component that connects to the
embedded activemq instance
CamelContext context = new DefaultCamelContext();
ActiveMQComponent activemqComponent =
ActiveMQComponent.activeMQComponent();
activemqComponent.setBrokerURL(String.format("vm://%s?create=false",
brokerName));
context.addComponent("activemq", activemqComponent);
// configure a route for processing requests
// - configure 3 concurrent consumers
// - simulate a lengthy processing time (12 seconds) for the first
request only
final String receiveUrl = "activemq:queue:dest?concurrentConsumers=3";
context.addRoutes(new RouteBuilder() {
@Override
public void configure() throws Exception {
from(receiveUrl)
//.threads(5)
.process(new Processor() {
@Override
public void process(Exchange exchange) throws Exception {
int requestId = exchange.getIn().getHeader("requestId",
Integer.class);
System.err.println("received request: " + requestId);
if (requestId == 0) {
Thread.sleep(12000L);
}
exchange.getOut().setBody("reply");
}
});
}
});
context.start();
System.err.println("listening on url: " + receiveUrl);
Thread.sleep(Long.MAX_VALUE);
}
}
// CLIENT
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 TestConcConsumersClient {
public static void main(String... args) throws Exception {
// set up camel context with activemq component
CamelContext context = new DefaultCamelContext();
ActiveMQComponent activemqComponent =
ActiveMQComponent.activeMQComponent();
activemqComponent.setConnectionFactory(new ActiveMQConnectionFactory(
null, null, "tcp://localhost:7001"));
context.addComponent("activemq", activemqComponent);
context.start();
final String sendUrl = "activemq:queue:dest";
System.err.println("sending to url: " + sendUrl);
// set up producer
// - repeatedly send requests, once every second
// - thread out each request so that if a response takes a long time,
// it doesn't delay the next request
final ProducerTemplate template = context.createProducerTemplate();
for (int requestId = 0; ; requestId++) {
final int finalRequestId = requestId;
new Thread() {
public void run() {
System.err.println("sending request: " + finalRequestId);
template.requestBodyAndHeader(sendUrl, "abc", "requestId",
finalRequestId);
System.err.println("received reply for request: " +
finalRequestId);
}
}.start();
Thread.sleep(1000L);
}
}
}
