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

java.lang.Object
  extended by org.springframework.jms.support.JmsAccessor
      extended by org.springframework.jms.support.destination.JmsDestinationAccessor
          extended by org.springframework.jms.listener.AbstractJmsListeningContainer
              extended by org.springframework.jms.listener.AbstractMessageListenerContainer
                  extended by org.apache.servicemix.jbi.cluster.requestor.AbstractJmsRequestorPool
                      extended by org.apache.servicemix.jbi.cluster.requestor.AbstractPollingRequestorPool
                          extended by 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


Copyright © 2005-2012 FuseSource. All Rights Reserved.