org.apache.servicemix.jbi.cluster.requestor
Class GenericJmsRequestorPool

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.apache.servicemix.jbi.cluster.requestor.AbstractJmsRequestorPool
                      org.apache.servicemix.jbi.cluster.requestor.AbstractPollingRequestorPool
                          org.apache.servicemix.jbi.cluster.requestor.GenericJmsRequestorPool

All Implemented Interfaces:

JmsRequestorPool, 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

public class GenericJmsRequestorPool
extends AbstractPollingRequestorPool

A pool of session / consumer / producer. Pool items are obtained using the getRequestor(String) method and released using Requestor#close() method.

Nested Class Summary

class

GenericJmsRequestorPool.AsyncMessageListenerInvoker

Nested classes/interfaces inherited from class org.apache.servicemix.jbi.cluster.requestor.AbstractPollingRequestorPool

AbstractPollingRequestorPool.Requestor, AbstractPollingRequestorPool.TransactionException

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

org.springframework.jms.listener.AbstractJmsListeningContainer.SharedConnectionNotInitializedException

Field Summary

static long

DEFAULT_RECEIVE_TIMEOUT The default receive timeout: 1000 ms = 1 second.

Fields inherited from class org.apache.servicemix.jbi.cluster.requestor.AbstractPollingRequestorPool

listener, parked, transacted, transactionManager

Fields inherited from class org.apache.servicemix.jbi.cluster.requestor.AbstractJmsRequestorPool

concurrentConsumers, DEFAULT_RECOVERY_INTERVAL, DEFAULT_THREAD_NAME_PREFIX, maxConcurrentConsumers

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

lifecycleMonitor, sharedConnectionMonitor

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

logger

Constructor Summary

GenericJmsRequestorPool()

Method Summary

protected AbstractPollingRequestorPool.Requestor	createRequestor(boolean consume)
void	doInitialize()
int	getActiveConsumerCount() Return the number of currently active consumers.
int	getIdleTaskExecutionLimit() Return the limit for idle executions of a receive task.
int	getMaxMessagesPerTask() Return the maximum number of messages to process in one task.
long	getReceiveTimeout()
AbstractPollingRequestorPool.Requestor	getRequestor(String id) Retrieve a parked session/consumer/producer triplet from the pool or a newly created one.
int	getScheduledConsumerCount() Return the number of currently scheduled consumers.
boolean	isSharedConnectionEnabled()
protected void	messageReceived(Object invoker, javax.jms.Session session) Tries scheduling a new invoker, since we know messages are coming in...
protected void	noMessageReceived(Object invoker, javax.jms.Session session) Marks the affected invoker as idle.
protected void	scheduleNewInvokerIfAppropriate() Schedule a new invoker, increasing the total number of scheduled invokers for this listener container, but only if the specified "maxConcurrentConsumers" limit has not been reached yet, and only if this listener container does not currently have idle invokers that are waiting for new messages already.
void	setIdleTaskExecutionLimit(int idleTaskExecutionLimit) Specify the limit for idle executions of a receive task, not having received any message within its execution.
void	setMaxMessagesPerTask(int maxMessagesPerTask) Specify the maximum number of messages to process in one task.
void	setMessageSelector(String selector)
void	setReceiveTimeout(long receiveTimeout) Set the timeout to use for receive calls, in milliseconds.
void	setSharedConnectionEnabled(boolean sharedConnectionEnabled)
protected boolean	sharedConnectionEnabled()

Methods inherited from class org.apache.servicemix.jbi.cluster.requestor.AbstractPollingRequestorPool

doShutdown, getListener, getTransacted, getTransactionManager, newRequestor, parkItem, resume, setListener, setTransacted, setTransactionManager

Methods inherited from class org.apache.servicemix.jbi.cluster.requestor.AbstractJmsRequestorPool

convertJmsAccessException, createConsumer, createConsumer, createDefaultTaskExecutor, createProducer, createProducer, doRescheduleTask, establishSharedConnection, getConcurrentConsumers, getMaxConcurrentConsumers, getTaskExecutor, handleListenerSetupFailure, initialize, recoverAfterListenerSetupFailure, refreshConnectionUntilSuccessful, refreshDestination, setConcurrentConsumers, setMaxConcurrentConsumers, setRecoveryInterval, setTaskExecutor, sleepInbetweenRecoveryAttempts, startSharedConnection, stopSharedConnection

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

checkMessageListener, commitIfNecessary, doExecuteListener, doInvokeListener, doInvokeListener, executeListener, getDefaultSubscriptionName, getDestination, getDestinationDescription, getDestinationName, getDurableSubscriptionName, getExceptionListener, getMessageListener, getMessageSelector, handleListenerException, invokeErrorHandler, invokeExceptionListener, invokeListener, isAcceptMessagesWhileStopping, isExposeListenerSession, isSessionLocallyTransacted, isSubscriptionDurable, rollbackIfNecessary, rollbackOnExceptionIfNecessary, setAcceptMessagesWhileStopping, setDestination, setDestinationName, setDurableSubscriptionName, setErrorHandler, setExceptionListener, setExposeListenerSession, setMessageListener, setSubscriptionDurable, validateConfiguration

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, start, stop, stop

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

getDestinationResolver, isPubSubDomain, resolveDestinationName, setDestinationResolver, setPubSubDomain

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

createConnection, createSession, getConnectionFactory, getSessionAcknowledgeMode, isClientAcknowledge, isSessionTransacted, setConnectionFactory, setSessionAcknowledgeMode, setSessionAcknowledgeModeName, setSessionTransacted

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.springframework.context.Lifecycle

isRunning, start, stop

Field Detail

DEFAULT_RECEIVE_TIMEOUT

public static final long DEFAULT_RECEIVE_TIMEOUT

The default receive timeout: 1000 ms = 1 second.

See Also:

Constant Field Values

Constructor Detail

GenericJmsRequestorPool

public GenericJmsRequestorPool()

Method Detail

setReceiveTimeout

public void setReceiveTimeout(long receiveTimeout)

Set the timeout to use for receive calls, in milliseconds. The default is 1000 ms, that is, 1 second.

NOTE: This value needs to be smaller than the transaction timeout used by the transaction manager (in the appropriate unit, of course). -1 indicates no timeout at all; however, this is only feasible if not running within a transaction manager.

See Also:

MessageConsumer.receive(long), MessageConsumer.receive(), #setTransactionTimeout

getReceiveTimeout

public long getReceiveTimeout()

isSharedConnectionEnabled

public boolean isSharedConnectionEnabled()

setSharedConnectionEnabled

public void setSharedConnectionEnabled(boolean sharedConnectionEnabled)

sharedConnectionEnabled

protected boolean sharedConnectionEnabled()

Specified by:

sharedConnectionEnabled in class org.springframework.jms.listener.AbstractJmsListeningContainer

createRequestor

protected AbstractPollingRequestorPool.Requestor createRequestor(boolean consume)
                                                          throws javax.jms.JMSException

Specified by:

createRequestor in class AbstractPollingRequestorPool

Throws:

javax.jms.JMSException

getScheduledConsumerCount

public final int getScheduledConsumerCount()

Return the number of currently scheduled consumers.

This number will always be inbetween "concurrentConsumers" and "maxConcurrentConsumers", but might be higher than "activeConsumerCount" (in case of some consumers being scheduled but not executed at the moment).

See Also:

AbstractJmsRequestorPool.getConcurrentConsumers(), AbstractJmsRequestorPool.getMaxConcurrentConsumers(), getActiveConsumerCount()

getActiveConsumerCount

public final int getActiveConsumerCount()

Return the number of currently active consumers.

This number will always be inbetween "concurrentConsumers" and "maxConcurrentConsumers", but might be lower than "scheduledConsumerCount". (in case of some consumers being scheduled but not executed at the moment).

See Also:

AbstractJmsRequestorPool.getConcurrentConsumers(), AbstractJmsRequestorPool.getMaxConcurrentConsumers(), getActiveConsumerCount()

setMaxMessagesPerTask

public void setMaxMessagesPerTask(int maxMessagesPerTask)

Specify the maximum number of messages to process in one task. More concretely, this limits the number of message reception attempts per task, which includes receive iterations that did not actually pick up a message until they hit their timeout (see the "receiveTimeout" property).

Default is unlimited (-1) in case of a standard TaskExecutor, reusing the original invoker threads until shutdown (at the expense of limited dynamic scheduling).

In case of a SchedulingTaskExecutor indicating a preference for short-lived tasks, the default is 10 instead. Specify a number of 10 to 100 messages to balance between rather long-lived and rather short-lived tasks here.

Long-lived tasks avoid frequent thread context switches through sticking with the same thread all the way through, while short-lived tasks allow thread pools to control the scheduling. Hence, thread pools will usually prefer short-lived tasks.

This setting can be modified at runtime, for example through JMX.

See Also:

AbstractJmsRequestorPool.setTaskExecutor(org.springframework.core.task.TaskExecutor), setReceiveTimeout(long), SchedulingTaskExecutor.prefersShortLivedTasks()

getMaxMessagesPerTask

public int getMaxMessagesPerTask()

Return the maximum number of messages to process in one task.

setIdleTaskExecutionLimit

public void setIdleTaskExecutionLimit(int idleTaskExecutionLimit)

Specify the limit for idle executions of a receive task, not having received any message within its execution. If this limit is reached, the task will shut down and leave receiving to other executing tasks.

Default is 1, closing idle resources early once a task didn't receive a message. This applies to dynamic scheduling only; see the "maxConcurrentConsumers" setting. The minimum number of consumers (see "concurrentConsumers") will be kept around until shutdown in any case.

Within each task execution, a number of message reception attempts (according to the "maxMessagesPerTask" setting) will each wait for an incoming message (according to the "receiveTimeout" setting). If all of those receive attempts in a given task return without a message, the task is considered idle with respect to received messages. Such a task may still be rescheduled; however, once it reached the specified "idleTaskExecutionLimit", it will shut down (in case of dynamic scaling).

Raise this limit if you encounter too frequent scaling up and down. With this limit being higher, an idle consumer will be kept around longer, avoiding the restart of a consumer once a new load of messages comes in. Alternatively, specify a higher "maxMessagesPerTask" and/or "receiveTimeout" value, which will also lead to idle consumers being kept around for a longer time (while also increasing the average execution time of each scheduled task).

This setting can be modified at runtime, for example through JMX.

See Also:

setMaxMessagesPerTask(int), setReceiveTimeout(long)

getIdleTaskExecutionLimit

public int getIdleTaskExecutionLimit()

Return the limit for idle executions of a receive task.

messageReceived

protected void messageReceived(Object invoker,
                               javax.jms.Session session)

Tries scheduling a new invoker, since we know messages are coming in...

See Also:

scheduleNewInvokerIfAppropriate()

noMessageReceived

protected void noMessageReceived(Object invoker,
                                 javax.jms.Session session)

Marks the affected invoker as idle.

scheduleNewInvokerIfAppropriate

protected void scheduleNewInvokerIfAppropriate()

Schedule a new invoker, increasing the total number of scheduled invokers for this listener container, but only if the specified "maxConcurrentConsumers" limit has not been reached yet, and only if this listener container does not currently have idle invokers that are waiting for new messages already.

Called once a message has been received, to scale up while processing the message in the invoker that originally received it.

See Also:

AbstractJmsRequestorPool.setTaskExecutor(org.springframework.core.task.TaskExecutor), AbstractJmsRequestorPool.getMaxConcurrentConsumers()

getRequestor

public AbstractPollingRequestorPool.Requestor getRequestor(String id)
                                                    throws javax.jms.JMSException

Retrieve a parked session/consumer/producer triplet from the pool or a newly created one. If an item has been previously parked with the same id, it will be unparked and returned, else, one will be obtained from the pool or created.

Parameters:

id - the parking id

Returns:

the parked item or a new one

Throws:

javax.jms.JMSException - if an error occur

doInitialize

public void doInitialize()
                  throws javax.jms.JMSException

Specified by:

doInitialize in class org.springframework.jms.listener.AbstractJmsListeningContainer

Throws:

javax.jms.JMSException

setMessageSelector

public void setMessageSelector(String selector)

Specified by:

setMessageSelector in interface JmsRequestorPool

Overrides:

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