Camel ActiveMQ Performance Tuning

Question

Situation

At present, we use some custom code on top of ActiveMQ libraries for JMS messaging. I have been looking at switching to Camel, for ease of use, ease of maintenance, and reliability.

Problem

With my present configuration, Camel's ActiveMQ implementation is substantially slower than our old implementation, both in terms of delay per message sent and received, and time taken to send and receive a large flood of messages. I've tried tweaking some configuration (e.g. maximum connections), to no avail.

Test Approach

I have two applications, one using our old implementation, one using a Camel implementation. Each application sends JMS messages to a topic on local ActiveMQ server, and also listens for messages on that topic. This is used to test two Scenarios: - Sending 100,000 messages to the topic in a loop, and seen how long it takes from start of sending to end of handling all of them. - Sending a message every 100 ms and measuring the delay (in ns) from sending to handling each message.

Question

Can I improve upon the implementation below, in terms of time sent to time processed for both floods of messages, and individual messages? Ideally, improvements would involve tweaking some config that I have missed, or suggesting a better way to do it, and not be too hacky. Explanations of improvements would be appreciated.

Edit: Now that I am sending messages asyncronously, I appear to have a concurrency issue. receivedCount does not reach 100,000. Looking at the ActiveMQ web interface, 100,000 messages are enqueued, and 100,000 dequeued, so it's probably a problem on the message processing side. I've altered receivedCount to be an AtomicInteger and added some logging to aid debugging. Could this be a problem with Camel itself (or the ActiveMQ components), or is there something wrong with the message processing code? As far as I can tell, only ~99,876 messages are making it through to floodProcessor.process.

Test Implementation

Edit: Updated with async sending and logging for concurrency issue.

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.camel.component.ActiveMQComponent;
import org.apache.activemq.pool.PooledConnectionFactory;
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.jms.JmsConfiguration;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.log4j.Logger;

public class CamelJmsTest{
    private static final Logger logger = Logger.getLogger(CamelJmsTest.class);

    private static final boolean flood = true;
    private static final int NUM_MESSAGES = 100000;

    private final CamelContext context;
    private final ProducerTemplate producerTemplate;

    private long timeSent = 0;

    private final AtomicInteger sendCount = new AtomicInteger(0);
    private final AtomicInteger receivedCount = new AtomicInteger(0);

    public CamelJmsTest() throws Exception {
        context = new DefaultCamelContext();

        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");

        PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(connectionFactory);

        JmsConfiguration jmsConfiguration = new JmsConfiguration(pooledConnectionFactory);
        logger.info(jmsConfiguration.isTransacted());

        ActiveMQComponent activeMQComponent = ActiveMQComponent.activeMQComponent();
        activeMQComponent.setConfiguration(jmsConfiguration);

        context.addComponent("activemq", activeMQComponent);

        RouteBuilder builder = new RouteBuilder() {
            @Override
            public void configure() {
                Processor floodProcessor = new Processor() {
                    @Override
                    public void process(Exchange exchange) throws Exception {
                        int newCount = receivedCount.incrementAndGet();

                        //TODO: Why doesn't newCount hit 100,000? Remove this logging once fixed
                        logger.info(newCount + ":" + exchange.getIn().getBody());

                        if(newCount == NUM_MESSAGES){
                            logger.info("all messages received at " + System.currentTimeMillis());
                        }
                    }
                };

                Processor spamProcessor = new Processor() {
                    @Override
                    public void process(Exchange exchange) throws Exception {
                        long delay = System.nanoTime() - timeSent;

                        logger.info("Message received: " + exchange.getIn().getBody(List.class) + " delay: " + delay);
                    }
                };

                from("activemq:topic:test?exchangePattern=InOnly")//.threads(8) // Having 8 threads processing appears to make things marginally worse
                    .choice()
                        .when(body().isInstanceOf(List.class)).process(flood ? floodProcessor : spamProcessor)
                    .otherwise().process(new Processor() {
                        @Override
                        public void process(Exchange exchange) throws Exception {
                            logger.info("Unknown message type received: " + exchange.getIn().getBody());
                        }
                    });
            }
        };

        context.addRoutes(builder);

        producerTemplate = context.createProducerTemplate();
        // For some reason, producerTemplate.asyncSendBody requires an Endpoint to be passed in, so the below is redundant:
//      producerTemplate.setDefaultEndpointUri("activemq:topic:test?exchangePattern=InOnly");
    }

    public void send(){
        int newCount = sendCount.incrementAndGet();
        producerTemplate.asyncSendBody("activemq:topic:test?exchangePattern=InOnly", Arrays.asList(newCount));
    }

    public void spam(){
        Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                timeSent = System.nanoTime();
                send();
            }
        }, 1000, 100, TimeUnit.MILLISECONDS);
    }

    public void flood(){
        logger.info("starting flood at " + System.currentTimeMillis());
        for (int i = 0; i < NUM_MESSAGES; i++) {
            send();
        }
        logger.info("flooded at " + System.currentTimeMillis());
    }

    public static void main(String... args) throws Exception {
        CamelJmsTest camelJmsTest = new CamelJmsTest();
        camelJmsTest.context.start();

        if(flood){
            camelJmsTest.flood();
        }else{
            camelJmsTest.spam();
        }
    }
}

Answer 1

It appears from your current JmsConfiguration that you are only consuming messages with a single thread. Was this intended?

If not, you need to set the concurrentConsumers property to something higher. This will create a threadpool of JMS listeners to service your destination.

Example:

JmsConfiguration config = new JmsConfiguration(pooledConnectionFactory);
config.setConcurrentConsumers(10);

This will create 10 JMS listener threads that will process messages concurrently from your queue.

EDIT:

For topics you can do something like this:

JmsConfiguration config = new JmsConfiguration(pooledConnectionFactory);
config.setConcurrentConsumers(1);
config.setMaxConcurrentConsumers(1);

And then in your route:

from("activemq:topic:test?exchangePattern=InOnly").threads(10)

Also, in ActiveMQ you can use a virtual destination. The virtual topic will act like a queue and then you can use the same concurrentConsumers method you would use for a normal queue.

Further Edit (For Sending):

You are currently doing a blocking send. You need to do producerTemplate.asyncSendBody().

---

Edit

I just built a project with your code and ran it. I set a breakpoint in your floodProcessor method and newCount is reaching 100,000. I think you may be getting thrown off by your logging and the fact that you are sending and receiving asynchronously. On my machine newCount hit 100,000 and the "all messages recieved" message was logged in well under 1 second after execution, but the program continued to log for another 45 seconds afterwards since it was buffered. You can see the effect of logging on how close your newCount number is to your body number by reducing the logging. I turned the logging to info, shutting off camel logging, and the two numbers matched at the end of the logging:

INFO  CamelJmsTest - 99996:[99996]
INFO  CamelJmsTest - 99997:[99997]
INFO  CamelJmsTest - 99998:[99998]
INFO  CamelJmsTest - 99999:[99999]
INFO  CamelJmsTest - 100000:[100000]
INFO  CamelJmsTest - all messages received at 1358778578422

Answer 2

I took over from the original poster in looking at this as part of another task, and found the problem with losing messages was actually in the ActiveMQ config.

We had a setting sendFailIfNoSpace=true, which was resulting in messages being dropped if we were sending fast enough to fill the publishers cache. Playing around with the policyEntry topic cache size I could vary the number of messages that disappeared with as much reliability as can be expected of such a race condition. Setting sendFailIfNoSpace=false (default), I could have any cache size I liked and never fail to receive all messages.

In theory sendFailIfNoSpace should throw a ResourceAllocationException when it drops a message, but that is either not happening(!) or is being ignored somehow. Also interesting is that our custom JMS wrapper code doesn't hit this problem despite running the throughput test faster than Camel. Maybe that code is faster in such a way that it means the publishing cache is being emptied faster, or else we are overriding sendFailIfNoSpace in the connection code somewhere that I haven't found yet.

On the question of speed, we have implemented all the suggestions mentioned here so far except for virtual destinations, but the Camel version test with 100K messages still runs in 16 seconds on my machine compared to 10 seconds for our own wrapper. As mentioned above, I have a sneaking suspicion that we are (implicitly or otherwise) overriding config somewhere in our wrapper, but I doubt it is anything that would cause that big a performance boost within ActiveMQ.

Virtual destinations as mentioned by gwithake might speed up this particular test, but most of the time with our real workloads it is not an appropriate solution.