Camel: Is it Possible to Implement Request/Reply & Competing Consumers with the Help of Asynchronous Processor?
apache-camel
Solution
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.
Problem
I'm struggling with the implementation of a Request/Reply exchange that should be processed concurrently by several Competing Consumers. I have one standalone Master-module that is responsible for producing a queue of tasks. And I have many Worker-modules that should consume messages from that queue concurrently. This is the Master-part of the Camel routing: ``` from("direct:start") .to("log:FROM.DIRECT?level=DEBUG") .split(body()).setHeader(CamelHeader.TASKS_BATCH_ID, simple("BATCH-1")) .setHeader(CamelHeader.TASK_TYPE, simple(TaskType.FETCH_INDEX)) .process(new Processor() { @Override public void process(Exchange exchange) throws Exception { EdgarFullIndexLocation location = exchange.getIn().getBody(EdgarFullIndexLocation.class); exchange.getIn().setBody(location.getId().toJson(), String.class); } }) .to("log:SPLIT?level=DEBUG") .setExchangePattern(ExchangePattern.InOut) .to("activemq:queue:tasksQueue?replyTo=completionsQueue" + //"&transactedInOut=true" + "&requestTimeout=" + Integer.MAX_VALUE + "&disableTimeToLive=true") .threads(10) .to("log:RESPONSE?level=DEBUG") .routeId(routeId); ``` This is the Worker part of the Camel routing, where I consume the queue: ``` from("activemq:queue:tasksQueue?asyncConsumer=true" + "&concurrentConsumers=10") .to("log:FROM.TASKS.QUEUE?level=DEBUG") .choice() .when(header(CamelHeader.TASK_TYPE).isEqualTo(TaskType.FETCH_INDEX)) .process(new FetchIndexTaskProcessor()) .otherwise() .to("log:UNKNOWN.TASK?level=DEBUG"); ``` Here the FetchIndexTaskProcessor implements AsyncProcessor: ``` public class FetchIndexTaskProcessor implements AsyncProcessor { @Override public void process(Exchange exchange) throws Exception {} @Override public boolean process(Exchange exchange, AsyncCallback callback) { FetchIndexTask task = new FetchIndexTask(exchange, callback); task.start(); return false; } } ``` Here the FetchIndexTask extends Thread. After start() the new thread is responsible for: - Dynamically adding a route. - Blocking until the exchange of that route completes. - Preparing a reply to the original exchange. - Calling `callback.done(false);` at the end. Everything works, except for the part of having Competing Consumers - it is always a single consumer at a time. I have tried many options like: - specifying a thread pool with `.threads(10)` in various places. - using endpoint options such as `asyncConsumer` and `concurrentConsumers` But it seems like I'm missing something important, and I can't seem to make it work in a concurrent fashion. What is the proper way of doing it?