How to improve efficiency with Camel?

Question

I have a problem of efficiency in my project which uses Camel with the Esper component.

I have several external datasources feeding information to camel endpoints. Each Camel endpoint that receives data transfers it to a route that processes it and then delivers it at an Esper endpoint.

The image below illustrates this behavior:

The efficiency problem is that all of this is done by a single Java thread. Thus if I have many sources, there is a huge bottleneck.

The following code accurately illustrates what is going on with the image:

public final void configure() throws OperationNotSupportedException{
    RouteDefinition route = from("xmpp://localhost:5222/?blablabla...");

    // apply some filter
    FilterDefinition filterDefinition = route.filter().method(...);

    // apply main processor
    ExpressionNode expressionNode = filterDefinition.process(...);


    // set destination
    expressionNode = filterDefinition.to("esper://session_X");
}

To fix this problem, I have to handle this situation with a pool of threads or using some sort of parallel processing. I cannot use patterns like multicast, recipient list, etc because all of those send the same message to multiple endpoints / clients, which is not the case in my examples.

A possible solution would be having 1 thread per each "Datasource endpoint -> Route -> Esper endpoint" combination, like the image bellow:

Another possible solution is to have 1 thread receive everything from the datasources, and then dispatch it to multiple threads handling the route processing together with the other endpoint:

PS: I am open to any other possible suggestions you may have.

To achieve one of these I have considered using the Camel SEDA component component, however, this one does not seem to allow me to have dynamic thread pools, because the concurrentConsumers property is static. Furthermore, I am not sure if I can use a SEDA endpoint at all, because I believe (although I am not completely sure) that the syntax for an endpoint like .to("seda:esper://session_X?concurrentConsumers=10") is invalid for Camel.

So, at this point I am quite lost and I don't know what to do: - Is SEDA the solution I am looking for? - If yes, how do I integrate it with the Esper endpoint given the syntax problem? - Are there any other solutions / Camel components that could fix my problem?

Answer 1

You must define a separate seda route that is distributing your message to the esper engine such as (using the fluent style):

public final void configure() throws OperationNotSupportedException{
    from("xmpp://localhost:5222/?blablabla...")
        .filter().method(...)
        .process(...)
        .to("seda:sub");

    from("seda:sub?concurrentConsumers=10)
     .to("esper://session_X");
 }

That said, seda should only be used if loosing messages is not a problem. Otherwise you should use a more robust protocol such as jms that allows to persist messages.

EDIT:

Beside seda, you could use threads(), where you could customize the threading behaviour by defining an ExecutorService:

public final void configure() throws OperationNotSupportedException{
    from("xmpp://localhost:5222/?blablabla...")
        .filter().method(...)
        .process(...)
        .threads()
        .executorService(Executors.newFixedThreadPool(2))
        .to("esper://session_X");
 }

If you using seda or threads(), you may loose transaction safety in case of failures. For this case or if you need to balance the workload to several remote hosts, you may use jms. More information about this solution is found here.