org.jadira.jms.container

Class BatchedMessageListenerContainer

java.lang.Object

org.springframework.jms.support.JmsAccessor

org.springframework.jms.support.destination.JmsDestinationAccessor

org.springframework.jms.listener.AbstractJmsListeningContainer

org.springframework.jms.listener.AbstractMessageListenerContainer

org.springframework.jms.listener.AbstractPollingMessageListenerContainer

org.springframework.jms.listener.DefaultMessageListenerContainer

org.jadira.jms.container.BatchedMessageListenerContainer

All Implemented Interfaces:

org.springframework.beans.factory.Aware, org.springframework.beans.factory.BeanNameAware, org.springframework.beans.factory.DisposableBean, org.springframework.beans.factory.InitializingBean, org.springframework.context.Lifecycle, org.springframework.context.Phased, org.springframework.context.SmartLifecycle, org.springframework.jms.listener.MessageListenerContainer

public class BatchedMessageListenerContainer
extends org.springframework.jms.listener.DefaultMessageListenerContainer

BatchedMessageListenerContainer is an extension of Spring's DefaultMessageListenerContainer which batches multiple message reads into each single transaction. Batching message reads typically provides significant enhancement to the throughput of message reads in message queue environments.

To use this class you must inject a transaction manager via AbstractPollingMessageListenerContainer.setTransactionManager(org.springframework.transaction.PlatformTransactionManager).

The class performs a blocking read for the first message read in any transaction. The blocking duration is determined by AbstractPollingMessageListenerContainer.setReceiveTimeout(long). Subsequent messages up to the configured setMaxMessagesPerTransaction(int) batch limit (which defaults to 150) are performed as non-blocking reads, with the bach completing as soon as the message queue cannot provide further messages.

Users of this class must handle rollback appropriately. A rollback triggered by failure processing a single message will cause all the messages in the transaction to rollback. It is recommended to design you message processing so that rollback only occurs for fatal, unexpected and unrecoverable errors such as a failure in the infrastructure. You should handle other errors by, for example, delivering messages directly to an error queue rather than throwing an exception. To assist in constructing this pattern, the AbstractMessageDriven POJO is also provided which provides the basic framework for implementing a MessageListener that is aligned with this contract.

The class contains an optional feature called RetryMitigation which is enabled by default. RetryMitigation prevents further messages being read in a batch if any message is identified as being one that is redelivered. When retryMitigation is enabled, any failure in processing will also trigger a pessimistic message mode. Once in pessimistic message mode messages are read one message at a time. This mode remains in place until either the number of messages in a batch multiplied by the number of concurrent consumers have been read since the last failure, or the queue cannot provide further messages (i.e. is empty). The aim of this feature is to reduce the likelihood of messages reaching the redelivery limit due to a bad message in the batch.

You can also configure the class to conclude any batch when a redelivered message is encountered (again the default behaviour). This feature complements RetryMitigation.

NB. Due to the design and structure of Spring's DefaultMessageListenerContainer and its superclasses, implementing this class must by necessity duplicate certain parts of DefaultMessageListenerContainer. Consequently, this class has been managed at a source code level as a derivative of DefaultMessageListenerContainer and copyright messages and attributions reflect this.

Author:

Juergen Hoeller was the original author of the DefaultMessageListenerContainer. The class was modified, extended and renamed to enable batching by Chris Pheby.

Nested Class Summary

Nested classes/interfaces inherited from class org.springframework.jms.listener.AbstractJmsListeningContainer

org.springframework.jms.listener.AbstractJmsListeningContainer.SharedConnectionNotInitializedException

Field Summary

Fields

Modifier and Type	Field and Description
static int	DEFAULT_BATCH_SIZE Default number of messages to read per transaction

Fields inherited from class org.springframework.jms.listener.DefaultMessageListenerContainer

CACHE_AUTO, CACHE_CONNECTION, CACHE_CONSUMER, CACHE_NONE, CACHE_SESSION, DEFAULT_RECOVERY_INTERVAL, DEFAULT_THREAD_NAME_PREFIX

Fields inherited from class org.springframework.jms.listener.AbstractPollingMessageListenerContainer

DEFAULT_RECEIVE_TIMEOUT

Fields inherited from class org.springframework.jms.listener.AbstractJmsListeningContainer

lifecycleMonitor, sharedConnectionMonitor

Fields inherited from class org.springframework.jms.support.destination.JmsDestinationAccessor

RECEIVE_TIMEOUT_INDEFINITE_WAIT, RECEIVE_TIMEOUT_NO_WAIT

Fields inherited from class org.springframework.jms.support.JmsAccessor

logger

Constructor Summary

Constructors

Constructor and Description
BatchedMessageListenerContainer() Create a new instance

Method Summary

Modifier and Type	Method and Description
protected void	commitIfNecessary(javax.jms.Session session, List<javax.jms.Message> messages) Variant of AbstractMessageListenerContainer.commitIfNecessary(Session, Message) that performs the activity for a batch of messages.
protected void	doExecuteListener(javax.jms.Session session, List<javax.jms.Message> messages) A batched variant of AbstractMessageListenerContainer.doExecuteListener(Session, Message).
protected boolean	doReceiveAndExecute(Object invoker, javax.jms.Session session, javax.jms.MessageConsumer consumer, org.springframework.transaction.TransactionStatus status)
int	getMaxMessagesPerTransaction() Get the configured maximum number of messages per transaction
boolean	isConcludeBatchOnRedeliveredMessage() True if seeing a redelivered message will conclude the current batch
boolean	isRetryMitigation() True if the instance attempts to mitigate the problems arising when messages in a batch are all retried when a poison message is encountered
protected javax.jms.Message	receiveMessageNoWait(javax.jms.MessageConsumer consumer) This is the BatchedMessageListenerContainer's equivalent to AbstractPollingMessageListenerContainer.receiveMessage(javax.jms.MessageConsumer).
void	setConcludeBatchOnRedeliveredMessage(boolean concludeBatchOnRedeliveredMessage) Enables or disables the ConcludeBatchOnRedeliveredMessage functionality
void	setMaxMessagesPerTransaction(int maxMessagesPerTransaction) Configures the maximum number of messages that can be read in a single transaction
void	setRetryMitigation(boolean retryMitigation) Enables or disables the RetryMitigation functionality
protected void	validateConfiguration()

Methods inherited from class org.springframework.jms.listener.DefaultMessageListenerContainer

applyBackOffTime, createDefaultTaskExecutor, doInitialize, doRescheduleTask, doShutdown, establishSharedConnection, getActiveConsumerCount, getCacheLevel, getConcurrentConsumers, getIdleConsumerLimit, getIdleTaskExecutionLimit, getMaxConcurrentConsumers, getMaxMessagesPerTask, getScheduledConsumerCount, handleListenerSetupFailure, initialize, isRecovering, isRegisteredWithDestination, messageReceived, noMessageReceived, recoverAfterListenerSetupFailure, refreshConnectionUntilSuccessful, refreshDestination, scheduleNewInvokerIfAppropriate, setBackOff, setCacheLevel, setCacheLevelName, setConcurrency, setConcurrentConsumers, setIdleConsumerLimit, setIdleTaskExecutionLimit, setMaxConcurrentConsumers, setMaxMessagesPerTask, setRecoveryInterval, setTaskExecutor, sharedConnectionEnabled, start, startSharedConnection, stop, stopSharedConnection

Methods inherited from class org.springframework.jms.listener.AbstractPollingMessageListenerContainer

createListenerConsumer, getConnection, getReceiveTimeout, getSession, getTransactionManager, isSessionLocallyTransacted, receiveAndExecute, receiveMessage, setReceiveTimeout, setSessionTransacted, setTransactionManager, setTransactionName, setTransactionTimeout, shouldCommitAfterNoMessageReceived

Methods inherited from class org.springframework.jms.listener.AbstractMessageListenerContainer

checkMessageListener, commitIfNecessary, createConsumer, doExecuteListener, doInvokeListener, doInvokeListener, executeListener, getDefaultSubscriptionName, getDestination, getDestinationDescription, getDestinationName, getDurableSubscriptionName, getErrorHandler, getExceptionListener, getMessageConverter, getMessageListener, getMessageSelector, getReplyQosSettings, getSubscriptionName, handleListenerException, invokeErrorHandler, invokeExceptionListener, invokeListener, isAcceptMessagesWhileStopping, isExposeListenerSession, isPubSubNoLocal, isReplyPubSubDomain, isSubscriptionDurable, isSubscriptionShared, rollbackIfNecessary, rollbackOnExceptionIfNecessary, setAcceptMessagesWhileStopping, setDestination, setDestinationName, setDurableSubscriptionName, setErrorHandler, setExceptionListener, setExposeListenerSession, setMessageConverter, setMessageListener, setMessageSelector, setPubSubNoLocal, setReplyPubSubDomain, setReplyQosSettings, setSubscriptionDurable, setSubscriptionName, setSubscriptionShared, setupMessageListener

Methods inherited from class org.springframework.jms.listener.AbstractJmsListeningContainer

afterPropertiesSet, createSharedConnection, destroy, doStart, doStop, getBeanName, getClientId, getPausedTaskCount, getPhase, getSharedConnection, isActive, isAutoStartup, isRunning, logRejectedTask, prepareSharedConnection, refreshSharedConnection, rescheduleTaskIfNecessary, resumePausedTasks, runningAllowed, setAutoStartup, setBeanName, setClientId, setPhase, shutdown, stop

Methods inherited from class org.springframework.jms.support.destination.JmsDestinationAccessor

getDestinationResolver, isPubSubDomain, receiveFromConsumer, resolveDestinationName, setDestinationResolver, setPubSubDomain

Methods inherited from class org.springframework.jms.support.JmsAccessor

convertJmsAccessException, createConnection, createSession, getConnectionFactory, getSessionAcknowledgeMode, isClientAcknowledge, isSessionTransacted, obtainConnectionFactory, setConnectionFactory, setSessionAcknowledgeMode, setSessionAcknowledgeModeName

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.springframework.jms.listener.MessageListenerContainer

getDestinationResolver, isPubSubDomain

Methods inherited from interface org.springframework.context.SmartLifecycle

isAutoStartup

Methods inherited from interface org.springframework.context.Lifecycle

isRunning, stop

Methods inherited from interface org.springframework.context.Phased

getPhase

Field Detail

DEFAULT_BATCH_SIZE

public static final int DEFAULT_BATCH_SIZE

Default number of messages to read per transaction

See Also:

Constant Field Values

Constructor Detail

BatchedMessageListenerContainer

public BatchedMessageListenerContainer()

Create a new instance

Method Detail

setMaxMessagesPerTransaction

public void setMaxMessagesPerTransaction(int maxMessagesPerTransaction)

Configures the maximum number of messages that can be read in a single transaction

Parameters:

maxMessagesPerTransaction - The requested maximum number of messages per transaction

getMaxMessagesPerTransaction

public int getMaxMessagesPerTransaction()

Get the configured maximum number of messages per transaction

Returns:

The maximum number of messages per transaction

isRetryMitigation

public boolean isRetryMitigation()

True if the instance attempts to mitigate the problems arising when messages in a batch are all retried when a poison message is encountered

Returns:

True if RetryMitigation is enabled

setRetryMitigation

public void setRetryMitigation(boolean retryMitigation)

Enables or disables the RetryMitigation functionality

Parameters:

retryMitigation - True if RetryMitigation is to be enabled, false otherwise

isConcludeBatchOnRedeliveredMessage

public boolean isConcludeBatchOnRedeliveredMessage()

True if seeing a redelivered message will conclude the current batch

Returns:

True if ConcludeBatchOnRedeliveredMessage is enabled

setConcludeBatchOnRedeliveredMessage

public void setConcludeBatchOnRedeliveredMessage(boolean concludeBatchOnRedeliveredMessage)

Enables or disables the ConcludeBatchOnRedeliveredMessage functionality

Parameters:

concludeBatchOnRedeliveredMessage - True if ConcludeBatchOnRedeliveredMessage is to be enabled, false otherwise

doReceiveAndExecute

protected boolean doReceiveAndExecute(Object invoker,
                                      javax.jms.Session session,
                                      javax.jms.MessageConsumer consumer,
                                      org.springframework.transaction.TransactionStatus status)
                               throws javax.jms.JMSException

Overrides:

doReceiveAndExecute in class org.springframework.jms.listener.AbstractPollingMessageListenerContainer

Throws:

javax.jms.JMSException

doExecuteListener

protected void doExecuteListener(javax.jms.Session session,
                                 List<javax.jms.Message> messages)
                          throws javax.jms.JMSException

A batched variant of AbstractMessageListenerContainer.doExecuteListener(Session, Message).

Parameters:

session - The session

messages - A list of messages

Throws:

javax.jms.JMSException - Indicates a problem during processing

commitIfNecessary

protected void commitIfNecessary(javax.jms.Session session,
                                 List<javax.jms.Message> messages)
                          throws javax.jms.JMSException

Variant of AbstractMessageListenerContainer.commitIfNecessary(Session, Message) that performs the activity for a batch of messages.

Parameters:

session - the JMS Session to commit

messages - the messages to acknowledge

Throws:

javax.jms.JMSException - in case of commit failure

validateConfiguration

protected void validateConfiguration()

Overrides:

validateConfiguration in class org.springframework.jms.listener.AbstractMessageListenerContainer

receiveMessageNoWait

protected javax.jms.Message receiveMessageNoWait(javax.jms.MessageConsumer consumer)
                                          throws javax.jms.JMSException

This is the BatchedMessageListenerContainer's equivalent to AbstractPollingMessageListenerContainer.receiveMessage(javax.jms.MessageConsumer). Does not block if no message is available.

Parameters:

consumer - The MessageConsumer to use

Returns:

The Message, if any

Throws:

javax.jms.JMSException - Indicates a problem occurred