How to: Implement a BatchMessageListenerContainer for bulk consuming a JMS queue

Question

I recently faced the need for a JMS consumer in Spring Integration - capable of consuming burst of high volume without stressing my target Oracle database with too many commits.

The DefaultMessageListenerContainer does not seem to support anything but message by message transactions.

I googled for solutions and found a couple - but the lot of them suffered from being implemented not by inheritance from DMLC but rather by cloning and modifying the original source code from same - making it vulnerable to break in case I later wish to move to a more recent version of spring-jms. Also the code being cloned referenced private properties of DMLC which consequently had to be left out. And to make it all work also a couple of interfaces and a custom message listener was needed. All in all I did not feel comfortable.

So - what to do?

Answer 1

Well - this is a simple and compact solution that is entirely based on a single class derived from DefaultMessageListenerContainer.

I have only tested with message-driven-channel-adapter and a ChainedTransactionManager though - since this is sort of the basic scenario when needing to do stuff like this.

This is the code:

package dk.itealisten.myservice.spring.components;

import org.springframework.jms.listener.DefaultMessageListenerContainer;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import java.util.ArrayList;
import java.util.Enumeration;

public class BatchMessageListenerContainer extends DefaultMessageListenerContainer {

    public static final int DEFAULT_BATCH_SIZE = 100;

    public int batchSize = DEFAULT_BATCH_SIZE;

    /**
     * Override the method receiveMessage to return an instance of BatchMessage - an inner class being declared further down.
     */
    @Override
    protected Message receiveMessage(MessageConsumer consumer) throws JMSException {
        BatchMessage batch = new BatchMessage();
        while (!batch.releaseAfterMessage(super.receiveMessage(consumer))) ;
        return batch.messages.size() == 0 ? null : batch;
    }

    /**
     * As BatchMessage implements the javax.jms.Message interface it fits perfectly into the DMLC - only caveat is that SimpleMessageConverter dont know how to convert it to a Spring Integration Message - but that can be helped.
     * As BatchMessage will only serve as a container to carry the actual javax.jms.Message's from DMLC to the MessageListener it need not provide meaningful implementations of the methods of the interface as long as they are there.
     */
    protected class BatchMessage implements Message {

        public ArrayList<Message> messages = new ArrayList<Message>();

        /**
         * Add message to the collection of messages and return true if the batch meets the criteria for releasing it to the MessageListener.
         */
        public boolean releaseAfterMessage(Message message) {
            if (message != null) {
                messages.add(message);
            }
            // Are we ready to release?
            return message == null || messages.size() >= batchSize;
        }

        // Below is only dummy-implementations of the abstract methods of javax.jms.Message

        @Override
        public String getJMSMessageID() throws JMSException {
            return null;
        }

        @Override
        public void setJMSMessageID(String s) throws JMSException {

        }

        @Override
        public long getJMSTimestamp() throws JMSException {
            return 0;
        }

        @Override
        public void setJMSTimestamp(long l) throws JMSException {

        }

        @Override
        public byte[] getJMSCorrelationIDAsBytes() throws JMSException {
            return new byte[0];
        }

        @Override
        public void setJMSCorrelationIDAsBytes(byte[] bytes) throws JMSException {

        }

        @Override
        public void setJMSCorrelationID(String s) throws JMSException {

        }

        @Override
        public String getJMSCorrelationID() throws JMSException {
            return null;
        }

        @Override
        public Destination getJMSReplyTo() throws JMSException {
            return null;
        }

        @Override
        public void setJMSReplyTo(Destination destination) throws JMSException {

        }

        @Override
        public Destination getJMSDestination() throws JMSException {
            return null;
        }

        @Override
        public void setJMSDestination(Destination destination) throws JMSException {

        }

        @Override
        public int getJMSDeliveryMode() throws JMSException {
            return 0;
        }

        @Override
        public void setJMSDeliveryMode(int i) throws JMSException {

        }

        @Override
        public boolean getJMSRedelivered() throws JMSException {
            return false;
        }

        @Override
        public void setJMSRedelivered(boolean b) throws JMSException {

        }

        @Override
        public String getJMSType() throws JMSException {
            return null;
        }

        @Override
        public void setJMSType(String s) throws JMSException {

        }

        @Override
        public long getJMSExpiration() throws JMSException {
            return 0;
        }

        @Override
        public void setJMSExpiration(long l) throws JMSException {

        }

        @Override
        public long getJMSDeliveryTime() throws JMSException {
            return 0;
        }

        @Override
        public void setJMSDeliveryTime(long l) throws JMSException {

        }

        @Override
        public int getJMSPriority() throws JMSException {
            return 0;
        }

        @Override
        public void setJMSPriority(int i) throws JMSException {

        }

        @Override
        public void clearProperties() throws JMSException {

        }

        @Override
        public boolean propertyExists(String s) throws JMSException {
            return false;
        }

        @Override
        public boolean getBooleanProperty(String s) throws JMSException {
            return false;
        }

        @Override
        public byte getByteProperty(String s) throws JMSException {
            return 0;
        }

        @Override
        public short getShortProperty(String s) throws JMSException {
            return 0;
        }

        @Override
        public int getIntProperty(String s) throws JMSException {
            return 0;
        }

        @Override
        public long getLongProperty(String s) throws JMSException {
            return 0;
        }

        @Override
        public float getFloatProperty(String s) throws JMSException {
            return 0;
        }

        @Override
        public double getDoubleProperty(String s) throws JMSException {
            return 0;
        }

        @Override
        public String getStringProperty(String s) throws JMSException {
            return null;
        }

        @Override
        public Object getObjectProperty(String s) throws JMSException {
            return null;
        }

        @Override
        public Enumeration getPropertyNames() throws JMSException {
            return null;
        }

        @Override
        public void setBooleanProperty(String s, boolean b) throws JMSException {

        }

        @Override
        public void setByteProperty(String s, byte b) throws JMSException {

        }

        @Override
        public void setShortProperty(String s, short i) throws JMSException {

        }

        @Override
        public void setIntProperty(String s, int i) throws JMSException {

        }

        @Override
        public void setLongProperty(String s, long l) throws JMSException {

        }

        @Override
        public void setFloatProperty(String s, float v) throws JMSException {

        }

        @Override
        public void setDoubleProperty(String s, double v) throws JMSException {

        }

        @Override
        public void setStringProperty(String s, String s1) throws JMSException {

        }

        @Override
        public void setObjectProperty(String s, Object o) throws JMSException {

        }

        @Override
        public void acknowledge() throws JMSException {

        }

        @Override
        public void clearBody() throws JMSException {

        }

        @Override
        public <T> T getBody(Class<T> aClass) throws JMSException {
            return null;
        }

        @Override
        public boolean isBodyAssignableTo(Class aClass) throws JMSException {
            return false;
        }
    }
}

Below is a sample showing how it could be used in a Spring application context:

<beans xmlns="http://www.springframework.org/schema/beans"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xmlns:int="http://www.springframework.org/schema/integration"
  xmlns:jms="http://www.springframework.org/schema/integration/jms"
  xmlns:p="http://www.springframework.org/schema/p"
  xsi:schemaLocation="
    http://www.springframework.org/schema/beans     
    http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
    http://www.springframework.org/schema/integration
    http://www.springframework.org/schema/integration/spring-integration-4.0.xsd
    http://www.springframework.org/schema/integration/jms
    http://www.springframework.org/schema/integration/jms/spring-integration-jms-4.0.xsd">

<!-- Plug in the BatchMessageListenerContainer in a message-driven-channel-adapter -->
<jms:message-driven-channel-adapter container-class="dk.itealisten.myservice.spring.components.BatchMessageListenerContainer"
  acknowledge="transacted"
  channel="from.mq"
  concurrent-consumers="5"
  max-concurrent-consumers="15"
  connection-factory="jmsConnectionFactory"
  transaction-manager="transactionManager"
  destination="my.mq.queue"
  />

<!-- Flow processing the BatchMessages being posted on the "from.mq" channel -->
<int:chain input-channel="from.mq" output-channel="nullChannel">
  <int:splitter expression="payload.messages" />
  <!-- This is where we deal with conversion to spring messages as the payload is now a single standard javax.jms.Message implementation -->
  <int:transformer ref="smc" method="fromMessage"/>
  <!-- And finally we persist -->
  <int:service-activator ref="jdbcPublisher" method="persist"/>
</int:chain>

<!-- Various supporting beans -->

<!-- A bean to handle the database persistance --> 
<bean id="jdbcPersistor" class="dk.itealisten.myservice.spring.components.JdbcPersistor" p:dataSource-ref="dataSource" />

<!-- A bean to handle the conversion that could not take place in the MessageListener as it don't know how to convert a BatchMessage -->
<bean id="smc" class="org.springframework.jms.support.converter.SimpleMessageConverter"/>

<!-- Transaction manager must make sure messages are committed outbound (JDBC) before cleaned up inbound (JMS). -->
<bean id="transactionManager" class="org.springframework.data.transaction.ChainedTransactionManager">
  <constructor-arg name="transactionManagers">
    <list>
      <bean class="org.springframework.jms.connection.JmsTransactionManager" p:connectionFactory-ref="jmsConnectionFactory" />
      <bean class="org.springframework.jdbc.datasource.DataSourceTransactionManager" p:dataSource-ref="dataSource" />
    </list>
  </constructor-arg>
</bean>