Synchronous Request Response with ActiveMQ and Spring

activemq-5-x-box-reflectionA few months ago I did a deep dive into Efficient Lightweight JMS with Spring and ActiveMQ where I focused on the details for asynchronous sending and receiving of messages. In an ideal world all messaging would be asynchronous. If you need a response then you should set up an asynchronous listener and either have enough state stored in the service or in the message that you can continue processing once the response has been received. However, when ideals lead to complexity, we have to make the decision of how much complexity can we tolerate for the performance we need. Sometimes it just makes more sense to use a synchronous request/response.

Request Response with JMS

ActiveMQ documentation actually has a pretty good overview of how request-response semantics work in JMS.

You might think at first that to implement request-response type operations in JMS that you should create a new consumer with a selector per request; or maybe create a new temporary queue per request.

Creating temporary destinations, consumers, producers and connections are all synchronous request-response operations with the broker and so should be avoided for processing each request as it results in lots of chat with the JMS broker.

The best way to implement request-response over JMS is to create a temporary queue and consumer per client on startup, set JMSReplyTo property on each message to the temporary queue and then use a correlationID on each message to correlate request messages to response messages. This avoids the overhead of creating and closing a consumer for each request (which is expensive). It also means you can share the same producer & consumer across many threads if you want (or pool them maybe).

This is a pretty good start but it requires some tweaking to work best in Spring. It also should be noted that Lingo and Camel are also suggested as options when using ActiveMQ. In my previous post I addressed why I don’t use either of these options. In short Camel is more power than is needed for basic messaging and Lingo is built on Jencks, neither of which have been updated in years.

Request Response in Spring

The first thing to notice is that its infeasible to create a consumer and temporary queue per client in Spring since pooling resources is required overcome the JmsTemplate gotchas. To get around this, I suggest using predefined request and response queues, removing the overhead of creating a temporary queue for each request/response. To allow for multiple consumers and producers on the same queue the JMSCorrelationId is used to correlated the request with its response message.

At this point I implemented the following naive solution:

@Component
public class Requestor {

    private static final class CorrelationIdPostProcessor implements MessagePostProcessor {

        private final String correlationId;

        public CorrelationIdPostProcessor( final String correlationId ) {
            this.correlationId = correlationId;
        }

        @Override
        public Message postProcessMessage( final Message msg ) throws JMSException {
            msg.setJMSCorrelationID( correlationId );
            return msg;
        }
    }

    private final JmsTemplate jmsTemplate;

    @Autowired
    public RequestGateway( JmsTemplate jmsTemplate ) {
        this.jmsTemplate = jmsTemplate;
    }

    public String request( final String request, String queue ) throws IOException {
        final String correlationId = UUID.randomUUID().toString();
        jmsTemplate.convertAndSend( queue+".request", request, new CorrelationIdPostProcessor( correlationId ) );
        return (String) jmsTemplate.receiveSelectedAndConvert( queue+".response", "JMSCorrelationID='" + correlationId + "'" );
    }
}

This worked for a while until the system started occasionally timing out when making a request against a particularly fast responding service. After some debugging it became apparent that the service was responding so quickly that the receive() call was not fully initialized, causing it to miss the message. Once it finished initializing, it would wait until the timeout and fail. Unfortunately, there is very little in the way of documentation for this and the best suggestion I could find still seemed to leave open the possibility for the race condition by creating the consumer after sending the message. Luckily, according to the JMS spec, a consumer becomes active as soon as it is created and, assuming the connection has been started, it will start consuming messages. This allows for the reordering of the method calls leading to the slightly more verbose but also more correct solution. (NOTE: Thanks to Aaron Korver for pointing out that ProducerConsumer needs to implement SessionCallback and that true needs to be passed to the JmsTemplate.execute() for the connection to be started.)

@Component
public class Requestor {

    private static final class ProducerConsumer implements SessionCallback<Message> {

        private static final int TIMEOUT = 5000;

        private final String msg;

        private final DestinationResolver destinationResolver;

        private final String queue;

        public ProducerConsumer( final String msg, String queue, final DestinationResolver destinationResolver ) {
            this.msg = msg;
            this.queue = queue;
            this.destinationResolver = destinationResolver;
        }

        public Message doInJms( final Session session ) throws JMSException {
            MessageConsumer consumer = null;
            MessageProducer producer = null;
            try {
                final String correlationId = UUID.randomUUID().toString();
                final Destination requestQueue =
                        destinationResolver.resolveDestinationName( session, queue+".request", false );
                final Destination replyQueue =
                        destinationResolver.resolveDestinationName( session, queue+".response", false );
                // Create the consumer first!
                consumer = session.createConsumer( replyQueue, "JMSCorrelationID = '" + correlationId + "'" );
                final TextMessage textMessage = session.createTextMessage( msg );
                textMessage.setJMSCorrelationID( correlationId );
                textMessage.setJMSReplyTo( replyQueue );
                // Send the request second!
                producer = session.createProducer( requestQueue );
                producer.send( requestQueue, textMessage );
                // Block on receiving the response with a timeout
                return consumer.receive( TIMEOUT );
            }
            finally {
                // Don't forget to close your resources
                JmsUtils.closeMessageConsumer( consumer );
                JmsUtils.closeMessageProducer( producer );
            }
        }
    }

    private final JmsTemplate jmsTemplate;

    @Autowired
    public Requestor( final JmsTemplate jmsTemplate ) {
        this.jmsTemplate = jmsTemplate;
     }

    public String request( final String request, String queue ) {
        // Must pass true as the second param to start the connection
        return (String) jmsTemplate.execute( new ProducerConsumer( msg, queue, jmsTemplate.getDestinationResolver() ), true );
    }
}

About Pooling

Once the request/response logic was correct it was time to load test. Almost instantly, memory usage exploded and the garbage collector started thrashing. Inspecting ActiveMQ with the Web Console showed that MessageConsumers were hanging around even though they were being explicitly closed using Spring’s own JmsUtils. Turns out, the CachingConnectionFactory‘s JavaDoc held the key to what was going on: “Note also that MessageConsumers obtained from a cached Session won’t get closed until the Session will eventually be removed from the pool.” However, if the MessageConsumers could be reused this wouldn’t be an issue. Unfortunately, CachingConnectionFactory caches MessageConsumers based on a hash key which contains the selector among other values. Obviously each request/response call, with its necessarily unique selector, was creating a new consumer that could never be reused. Luckily ActiveMQ provides a PooledConnectionFactory which does not cache MessageConsumers and switching to it fixed the problem instantly. However, this means that each request/response requires a new MessageConsumer to be created. This is adds overhead but its the price that must be payed to do synchronous request/response.

Efficient Lightweight JMS with Spring and ActiveMQ

Spring

© Darwin Bell

Asynchronicity, its the number one design principal for highly scalable systems, and for Java that means JMS, which in turn means ActiveMQ. But how do I use JMS efficiently? One can quickly  become overwhelmed with talk of containers, frameworks, and a plethora of options, most of which are outdated. So lets pick it apart.

Frameworks

The ActiveMQ documentation makes mention of two frameworks; Camel and Spring. The decision here comes down to simplicity vs functionality. Camel supports an immense amount of Enterprise Integration Patterns that can greatly simplify integrating a variety of services and orchestrating complicated message flows between components. Its certainly a best of breed if your system requires such functionality. However, if you are looking for simplicity and support for the basic best practices then Spring has the upper hand. For me, simplicity wins out any day of the week.

JCA (Use It Or Loose It)

Reading through ActiveMQ’s spring support one is instantly introduced to the idea of a JCA container and ActiveMQ’s various proxies and adaptors for working inside of one. However, this is all a red herring. JCA is part of the EJB specification and as with most of the EJB specification, Spring doesn’t support it. Then there is a mention of Jencks, a “lightweight JCA container for Spring”, which was spun off of ActiveMQ’s JCA container. At first this seems like the ideal solution, but let me stop you there.  Jencks was last updated on January 3rd 2007. At that time ActiveMQ was at version 4.1.x and Spring was at version 2.0.x and things have come a long way, a very long way. Even trying to get Jencks from the maven repository fails due to dependencies on ActiveMQ 4.1.x jars that no longer exist. The simple fact is there are better and simpler ways to ensure resource caching.

Sending Messages

The core of Spring’s message sending architecture is the JmsTemplate. In typical Spring template fashion, the JmsTemplate abstracts away all the cruft of opening and closing sessions and producers so all the application developer needs to worry about is the actual business logic. However, ActiveMQ is quick to point out the JmsTemplate gotchas, mostly that JmsTemplate is designed to open and close the session and producer on each call. To prevent this from absolutely destroying the messaging performance the documentation recommends using ActiveMQ’s PooledConnectionFactory which caches the sessions and message producers. However this too is outdated. Starting with version 2.5.3, Spring started shipping its own CachingConnectionFactory which I believe to be the preferred caching method. (UPDATE: In my more recent post, I talk about when you might want to use PooledConnectionFactory.) However, there is one catch to point out. By default the CachingConnectionFactory only caches one session which the javadoc claims to be sufficient for low concurrency situations. By contrast, the PooledConnectionFactory defaults to 500. As with most settings of this type, some amount of experimentation is probably in order. I’ve started with 100 which seems like a good compromise.

Receiving Messages

As you may have noticed, the JmsTemplate gotchas strongly discourages using the recieve() call on the JmsTemplate, again, since there is no pooling of sessions and consumers.  Moreover, all calls on the JmsTemplate are synchronous which means the calling thread will block until the method returns. This is fine when using JmsTemplate to send messages since the method returns almost instantly. However, when using the recieve() call, the thread will block until a message is received, which has a huge impact on performance. Unfortunately, neither the JmsTemplate gotchas nor the spring support documentation mentions the simple Spring solution for these problems. In fact they both recommend using Jencks, which we already debunked. The actual solution, using the DefaultMessageListenerContainer, is buried in the how do I use JMS efficiently documentation. The DefaultMessageListenerContainer allows for the asynchronous receipt of messages as well as caching sessions and message consumers. Even more interesting, the DefaultMessageListenerContainer can dynamically grow and shrink the number of listeners based on message volume. In short, this is why we can completely ignore JCA.

Putting It All Together

Spring Context XML

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:amq="http://activemq.apache.org/schema/core"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.2.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd
http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-2.5.xsd">

<!-- enables annotation based configuration -->
<context:annotation-config />
<!-- scans for annotated classes in the com.company package -->
<context:component-scan base-package="com.company"/>
<!-- allows for ${} replacement in the spring xml configuration from the system.properties file on the classpath -->
<context:property-placeholder location="classpath:system.properties"/>
<!-- creates an activemq connection factory using the amq namespace -->
<amq:connectionFactory id="amqConnectionFactory" brokerURL="${jms.url}" userName="${jms.username}" password="${jms.password}" />
<!-- CachingConnectionFactory Definition, sessionCacheSize property is the number of sessions to cache -->
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
    <constructor-arg ref="amqConnectionFactory" />
    <property name="exceptionListener" ref="jmsExceptionListener" />
    <property name="sessionCacheSize" value="100" />
</bean>
<!-- JmsTemplate Definition -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
   <constructor-arg ref="connectionFactory"/>
</bean>
<!-- listener container definition using the jms namespace, concurrency is the max number of concurrent listeners that can be started -->
<jms:listener-container concurrency="10" >
    <jms:listener id="QueueListener" destination="Queue.Name" ref="queueListener" />
</jms:listener-container>
</beans>

There are two things to notice here. First, I’ve added the amq and jms namespaces to the opening beans tag. Second, I’m using the Spring 2.5 annotation based configuration. By using the annotation based configuration I can simply add @Component annotation to my Java classes instead of having to specify them in the spring context xml explicitly. Additionally, I can add @Autowired on my constructors to have objects such as JmsTemplate automatically wired into my objects.

QueueSender

package com.company;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Component;

@Component
public class QueueSender
{
    private final JmsTemplate jmsTemplate;

    @Autowired
    public QueueSender( final JmsTemplate jmsTemplate )
    {
        this.jmsTemplate = jmsTemplate;
    }

    public void send( final String message )
    {
        jmsTemplate.convertAndSend( "Queue.Name", message );
    }
}

Queue Listener


package com.company;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

import org.springframework.stereotype.Component;

@Component
public class QueueListener implements MessageListener
{
    public void onMessage( final Message message )
    {
        if ( message instanceof TextMessage )
        {
            final TextMessage textMessage = (TextMessage) message;
            try
            {
                System.out.println( textMessage.getText() );
            }
            catch (final JMSException e)
            {
                e.printStackTrace();
            }
        }
    }
}

JmsExceptionListener

package com.company;

import javax.jms.ExceptionListener;
import javax.jms.JMSException;

import org.springframework.stereotype.Component;

@Component
public class JmsExceptionListener implements ExceptionListener
{
    public void onException( final JMSException e )
    {
        e.printStackTrace();
    }
}

Update

I have finally updated the wiki at activemq.apache.org. The following pages now recommend using MessageListenerContainers and JmsTemplate with a Pooling ConnectionFactory instead of JCA and Jencks.
Follow

Get every new post delivered to your Inbox.