public class DefaultMessageListenerContainer extends AbstractPollingMessageListenerContainer
MessageConsumer.receive()
calls that also allow for
transactional reception of messages (registering them with XA transactions).
Designed to work in a native JMS environment as well as in a J2EE environment,
with only minimal differences in configuration.
This is a simple but nevertheless powerful form of message listener container.
On startup, it obtains a fixed number of JMS Sessions to invoke the listener,
and optionally allows for dynamic adaptation at runtime (up to a maximum number).
Like SimpleMessageListenerContainer
, its main advantage is its low level
of runtime complexity, in particular the minimal requirements on the JMS provider:
not even the JMS ServerSessionPool
facility is required. Beyond that, it is
fully self-recovering in case the broker is temporarily unavailable, and allows
for stops/restarts as well as runtime changes to its configuration.
Actual MessageListener
execution happens in asynchronous work units which are
created through Spring's TaskExecutor
abstraction. By default, the specified number of invoker tasks will be created
on startup, according to the "concurrentConsumers"
setting. Specify an alternative TaskExecutor
to integrate with an existing
thread pool facility (such as a J2EE server's), for example using a
CommonJ WorkManager
.
With a native JMS setup, each of those listener threads is going to use a
cached JMS Session
and MessageConsumer
(only refreshed in case
of failure), using the JMS provider's resources as efficiently as possible.
Message reception and listener execution can automatically be wrapped
in transactions by passing a Spring
PlatformTransactionManager
into the
"transactionManager"
property. This will usually
be a JtaTransactionManager
in a
J2EE environment, in combination with a JTA-aware JMS ConnectionFactory
obtained from JNDI (check your J2EE server's documentation). Note that this
listener container will automatically reobtain all JMS handles for each transaction
in case an external transaction manager is specified, for compatibility with
all J2EE servers (in particular JBoss). This non-caching behavior can be
overridden through the "cacheLevel"
/
"cacheLevelName"
property, enforcing caching of
the Connection
(or also Session
and MessageConsumer
)
even if an external transaction manager is involved.
Dynamic scaling of the number of concurrent invokers can be activated
by specifying a "maxConcurrentConsumers"
value that is higher than the "concurrentConsumers"
value. Since the latter's default is 1, you can also simply specify a
"maxConcurrentConsumers" of e.g. 5, which will lead to dynamic scaling up to
5 concurrent consumers in case of increasing message load, as well as dynamic
shrinking back to the standard number of consumers once the load decreases.
Consider adapting the "idleTaskExecutionLimit"
setting to control the lifespan of each new task, to avoid frequent scaling up
and down, in particular if the ConnectionFactory
does not pool JMS
Sessions
and/or the TaskExecutor
does not pool threads (check
your configuration!). Note that dynamic scaling only really makes sense for a
queue in the first place; for a topic, you will typically stick with the default
number of 1 consumer, otherwise you'd receive the same message multiple times on
the same node.
Note: Don't use Spring's CachingConnectionFactory
in combination with dynamic scaling. Ideally, don't use it with a message
listener container at all, since it is generally preferable to let the
listener container itself handle appropriate caching within its lifecycle.
Also, stopping and restarting a listener container will only work with an
independent, locally cached Connection - not with an externally cached one.
It is strongly recommended to either set "sessionTransacted"
to "true" or specify an external "transactionManager"
. See the AbstractMessageListenerContainer
javadoc for details on acknowledge modes and native transaction options, as
well as the AbstractPollingMessageListenerContainer
javadoc for details
on configuring an external transaction manager. Note that for the default
"AUTO_ACKNOWLEDGE" mode, this container applies automatic message acknowledgment
before listener execution, with no redelivery in case of an exception.
AbstractJmsListeningContainer.SharedConnectionNotInitializedException
Modifier and Type | Field and Description |
---|---|
static int |
CACHE_AUTO
Constant that indicates automatic choice of an appropriate caching level
(depending on the transaction management strategy).
|
static int |
CACHE_CONNECTION
Constant that indicates to cache a shared JMS
Connection for each
listener thread. |
static int |
CACHE_CONSUMER
Constant that indicates to cache a shared JMS
Connection , a JMS
Session , and a JMS MessageConsumer for each listener thread. |
static int |
CACHE_NONE
Constant that indicates to cache no JMS resources at all.
|
static int |
CACHE_SESSION
Constant that indicates to cache a shared JMS
Connection and a JMS
Session for each listener thread. |
static long |
DEFAULT_RECOVERY_INTERVAL
The default recovery interval: 5000 ms = 5 seconds.
|
static String |
DEFAULT_THREAD_NAME_PREFIX
Default thread name prefix: "DefaultMessageListenerContainer-".
|
DEFAULT_RECEIVE_TIMEOUT
lifecycleMonitor, sharedConnectionMonitor
logger
Constructor and Description |
---|
DefaultMessageListenerContainer() |
Modifier and Type | Method and Description |
---|---|
protected TaskExecutor |
createDefaultTaskExecutor()
Create a default TaskExecutor.
|
protected void |
doInitialize()
Creates the specified number of concurrent consumers,
in the form of a JMS Session plus associated MessageConsumer
running in a separate thread.
|
protected void |
doRescheduleTask(Object task)
Re-executes the given task via this listener container's TaskExecutor.
|
protected void |
doShutdown()
Destroy the registered JMS Sessions and associated MessageConsumers.
|
protected void |
establishSharedConnection()
Overridden to accept a failure in the initial setup - leaving it up to the
asynchronous invokers to establish the shared Connection on first access.
|
int |
getActiveConsumerCount()
Return the number of currently active consumers.
|
int |
getCacheLevel()
Return the level of caching that this listener container is allowed to apply.
|
int |
getConcurrentConsumers()
Return the "concurrentConsumer" setting.
|
int |
getIdleConsumerLimit()
Return the limit for the number of idle consumers.
|
int |
getIdleTaskExecutionLimit()
Return the limit for idle executions of a consumer task.
|
int |
getMaxConcurrentConsumers()
Return the "maxConcurrentConsumer" setting.
|
int |
getMaxMessagesPerTask()
Return the maximum number of messages to process in one task.
|
int |
getScheduledConsumerCount()
Return the number of currently scheduled consumers.
|
protected void |
handleListenerSetupFailure(Throwable ex,
boolean alreadyRecovered)
Handle the given exception that arose during setup of a listener.
|
void |
initialize()
Initialize this container.
|
boolean |
isRecovering()
Return whether this listener container is currently in a recovery attempt.
|
boolean |
isRegisteredWithDestination()
Return whether at least one consumer has entered a fixed registration with the
target destination.
|
protected void |
messageReceived(Object invoker,
Session session)
Tries scheduling a new invoker, since we know messages are coming in...
|
protected void |
noMessageReceived(Object invoker,
Session session)
Marks the affected invoker as idle.
|
protected void |
recoverAfterListenerSetupFailure()
Recover this listener container after a listener failed to set itself up,
for example re-establishing the underlying Connection.
|
protected void |
refreshConnectionUntilSuccessful()
Refresh the underlying Connection, not returning before an attempt has been
successful.
|
protected void |
refreshDestination()
Refresh the JMS destination that this listener container operates on.
|
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 the specified "idleConsumerLimit" has not been reached either.
|
void |
setCacheLevel(int cacheLevel)
Specify the level of caching that this listener container is allowed to apply.
|
void |
setCacheLevelName(String constantName)
Specify the level of caching that this listener container is allowed to apply,
in the form of the name of the corresponding constant: e.g.
|
void |
setConcurrency(String concurrency)
Specify concurrency limits via a "lower-upper" String, e.g.
|
void |
setConcurrentConsumers(int concurrentConsumers)
Specify the number of concurrent consumers to create.
|
void |
setIdleConsumerLimit(int idleConsumerLimit)
Specify the limit for the number of consumers that are allowed to be idle
at any given time.
|
void |
setIdleTaskExecutionLimit(int idleTaskExecutionLimit)
Specify the limit for idle executions of a consumer task, not having
received any message within its execution.
|
void |
setMaxConcurrentConsumers(int maxConcurrentConsumers)
Specify the maximum number of concurrent consumers to create.
|
void |
setMaxMessagesPerTask(int maxMessagesPerTask)
Specify the maximum number of messages to process in one task.
|
void |
setRecoveryInterval(long recoveryInterval)
Specify the interval between recovery attempts, in milliseconds.
|
void |
setTaskExecutor(Executor taskExecutor)
Set the Spring
TaskExecutor to use for running the listener threads. |
protected boolean |
sharedConnectionEnabled()
Use a shared JMS Connection depending on the "cacheLevel" setting.
|
protected void |
sleepInbetweenRecoveryAttempts()
Sleep according to the specified recovery interval.
|
void |
start()
Overridden to reset the stop callback, if any.
|
protected void |
startSharedConnection()
This implementations proceeds even after an exception thrown from
Connection.start() , relying on listeners to perform
appropriate recovery. |
void |
stop(Runnable callback)
Stop this listener container, invoking the specific callback
once all listener processing has actually stopped.
|
protected void |
stopSharedConnection()
This implementations proceeds even after an exception thrown from
Connection.stop() , relying on listeners to perform
appropriate recovery after a restart. |
createConsumer, createListenerConsumer, doReceiveAndExecute, getConnection, getSession, getTransactionManager, isPubSubNoLocal, isSessionLocallyTransacted, receiveAndExecute, receiveMessage, setPubSubNoLocal, setReceiveTimeout, setSessionTransacted, setTransactionManager, setTransactionName, setTransactionTimeout, shouldCommitAfterNoMessageReceived
checkMessageListener, commitIfNecessary, doExecuteListener, doInvokeListener, doInvokeListener, executeListener, getDefaultSubscriptionName, getDestination, getDestinationDescription, getDestinationName, getDurableSubscriptionName, getExceptionListener, getMessageListener, getMessageSelector, handleListenerException, invokeErrorHandler, invokeExceptionListener, invokeListener, isAcceptMessagesWhileStopping, isExposeListenerSession, isSubscriptionDurable, rollbackIfNecessary, rollbackOnExceptionIfNecessary, setAcceptMessagesWhileStopping, setDestination, setDestinationName, setDurableSubscriptionName, setErrorHandler, setExceptionListener, setExposeListenerSession, setMessageListener, setMessageSelector, setSubscriptionDurable, validateConfiguration
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
getDestinationResolver, isPubSubDomain, resolveDestinationName, setDestinationResolver, setPubSubDomain
convertJmsAccessException, createConnection, createSession, getConnectionFactory, getSessionAcknowledgeMode, isClientAcknowledge, isSessionTransacted, setConnectionFactory, setSessionAcknowledgeMode, setSessionAcknowledgeModeName
public static final String DEFAULT_THREAD_NAME_PREFIX
public static final long DEFAULT_RECOVERY_INTERVAL
public static final int CACHE_NONE
setCacheLevel(int)
,
Constant Field Valuespublic static final int CACHE_CONNECTION
Connection
for each
listener thread.setCacheLevel(int)
,
Constant Field Valuespublic static final int CACHE_SESSION
Connection
and a JMS
Session
for each listener thread.setCacheLevel(int)
,
Constant Field Valuespublic static final int CACHE_CONSUMER
Connection
, a JMS
Session
, and a JMS MessageConsumer for each listener thread.setCacheLevel(int)
,
Constant Field Valuespublic static final int CACHE_AUTO
setCacheLevel(int)
,
Constant Field Valuespublic void setTaskExecutor(Executor taskExecutor)
TaskExecutor
to use for running the listener threads.
Default is a SimpleAsyncTaskExecutor
,
starting up a number of new threads, according to the specified number
of concurrent consumers.
Specify an alternative TaskExecutor
for integration with an existing
thread pool. Note that this really only adds value if the threads are
managed in a specific fashion, for example within a J2EE environment.
A plain thread pool does not add much value, as this listener container
will occupy a number of threads for its entire lifetime.
public void setRecoveryInterval(long recoveryInterval)
public void setCacheLevelName(String constantName) throws IllegalArgumentException
IllegalArgumentException
setCacheLevel(int)
public void setCacheLevel(int cacheLevel)
Default is CACHE_NONE
if an external transaction manager has been specified
(to reobtain all resources freshly within the scope of the external transaction),
and CACHE_CONSUMER
otherwise (operating with local JMS resources).
Some J2EE servers only register their JMS resources with an ongoing XA
transaction in case of a freshly obtained JMS Connection
and Session
,
which is why this listener container by default does not cache any of those.
However, if you want to optimize for a specific server, consider switching
this setting to at least CACHE_CONNECTION
or CACHE_SESSION
even in conjunction with an external transaction manager.
public int getCacheLevel()
public void setConcurrency(String concurrency)
This listener container will always hold on to the minimum number of consumers
(setConcurrentConsumers(int)
) and will slowly scale up to the maximum number
of consumers setMaxConcurrentConsumers(int)
in case of increasing load.
public void setConcurrentConsumers(int concurrentConsumers)
Specifying a higher value for this setting will increase the standard level of scheduled concurrent consumers at runtime: This is effectively the minimum number of concurrent consumers which will be scheduled at any given time. This is a static setting; for dynamic scaling, consider specifying the "maxConcurrentConsumers" setting instead.
Raising the number of concurrent consumers is recommendable in order to scale the consumption of messages coming in from a queue. However, note that any ordering guarantees are lost once multiple consumers are registered. In general, stick with 1 consumer for low-volume queues.
Do not raise the number of concurrent consumers for a topic, unless vendor-specific setup measures clearly allow for it. With regular setup, this would lead to concurrent consumption of the same message, which is hardly ever desirable.
This setting can be modified at runtime, for example through JMX.
setMaxConcurrentConsumers(int)
public final int getConcurrentConsumers()
This returns the currently configured "concurrentConsumers" value; the number of currently scheduled/active consumers might differ.
public void setMaxConcurrentConsumers(int maxConcurrentConsumers)
If this setting is higher than "concurrentConsumers", the listener container will dynamically schedule new consumers at runtime, provided that enough incoming messages are encountered. Once the load goes down again, the number of consumers will be reduced to the standard level ("concurrentConsumers") again.
Raising the number of concurrent consumers is recommendable in order to scale the consumption of messages coming in from a queue. However, note that any ordering guarantees are lost once multiple consumers are registered. In general, stick with 1 consumer for low-volume queues.
Do not raise the number of concurrent consumers for a topic, unless vendor-specific setup measures clearly allow for it. With regular setup, this would lead to concurrent consumption of the same message, which is hardly ever desirable.
This setting can be modified at runtime, for example through JMX.
setConcurrentConsumers(int)
public final int getMaxConcurrentConsumers()
This returns the currently configured "maxConcurrentConsumers" value; the number of currently scheduled/active consumers might differ.
public void setMaxMessagesPerTask(int maxMessagesPerTask)
"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.
public final int getMaxMessagesPerTask()
public void setIdleConsumerLimit(int idleConsumerLimit)
This limit is used by the scheduleNewInvokerIfAppropriate()
method
to determine if a new invoker should be created. Increasing the limit causes
invokers to be created more aggressively. This can be useful to ramp up the
number of invokers faster.
The default is 1, only scheduling a new invoker (which is likely to be idle initially) if none of the existing invokers is currently idle.
public final int getIdleConsumerLimit()
public void setIdleTaskExecutionLimit(int idleTaskExecutionLimit)
The 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.
public final int getIdleTaskExecutionLimit()
public void initialize()
AbstractJmsListeningContainer
Creates a JMS Connection, starts the Connection
(if "autoStartup"
hasn't been turned off),
and calls AbstractJmsListeningContainer.doInitialize()
.
initialize
in class AbstractPollingMessageListenerContainer
protected void doInitialize() throws JMSException
doInitialize
in class AbstractJmsListeningContainer
JMSException
- if registration failedscheduleNewInvoker()
,
setTaskExecutor(java.util.concurrent.Executor)
protected void doShutdown() throws JMSException
doShutdown
in class AbstractJmsListeningContainer
JMSException
- if shutdown failedAbstractJmsListeningContainer.shutdown()
public void start() throws JmsException
start
in interface Lifecycle
start
in class AbstractJmsListeningContainer
JmsException
- if starting failedAbstractJmsListeningContainer.doStart()
public void stop(Runnable callback) throws JmsException
Note: Further stop(runnable)
calls (before processing
has actually stopped) will override the specified callback. Only the
latest specified callback will be invoked.
If a subsequent start()
call restarts the listener container
before it has fully stopped, the callback will not get invoked at all.
stop
in interface SmartLifecycle
stop
in class AbstractJmsListeningContainer
callback
- the callback to invoke once listener processing
has fully stoppedJmsException
- if stopping failedAbstractJmsListeningContainer.stop()
public final int getScheduledConsumerCount()
This number will always be between "concurrentConsumers" and "maxConcurrentConsumers", but might be higher than "activeConsumerCount" (in case some consumers are scheduled but not executing at the moment).
public final int getActiveConsumerCount()
This number will always be between "concurrentConsumers" and "maxConcurrentConsumers", but might be lower than "scheduledConsumerCount" (in case some consumers are scheduled but not executing at the moment).
public boolean isRegisteredWithDestination()
This method may be polled after a start()
call, until asynchronous
registration of consumers has happened which is when the method will start returning
true
– provided that the listener container ever actually establishes
a fixed registration. It will then keep returning true
until shutdown,
since the container will hold on to at least one consumer registration thereafter.
Note that a listener container is not bound to having a fixed registration in
the first place. It may also keep recreating consumers for every invoker execution.
This particularly depends on the cache level
setting:
only CACHE_CONSUMER
will lead to a fixed registration.
protected TaskExecutor createDefaultTaskExecutor()
The default implementation builds a SimpleAsyncTaskExecutor
with the specified bean name (or the class name, if no bean name specified) as thread name prefix.
protected final boolean sharedConnectionEnabled()
sharedConnectionEnabled
in class AbstractJmsListeningContainer
setCacheLevel(int)
,
CACHE_CONNECTION
protected void doRescheduleTask(Object task)
doRescheduleTask
in class AbstractJmsListeningContainer
task
- the task object to reschedulesetTaskExecutor(java.util.concurrent.Executor)
protected void messageReceived(Object invoker, Session session)
messageReceived
in class AbstractPollingMessageListenerContainer
invoker
- the invoker object (passed through)session
- the receiving JMS SessionscheduleNewInvokerIfAppropriate()
protected void noMessageReceived(Object invoker, Session session)
noMessageReceived
in class AbstractPollingMessageListenerContainer
invoker
- the invoker object (passed through)session
- the receiving JMS Sessionprotected void scheduleNewInvokerIfAppropriate()
Called once a message has been received, in order to scale up while processing the message in the invoker that originally received it.
protected void establishSharedConnection()
establishSharedConnection
in class AbstractJmsListeningContainer
refreshConnectionUntilSuccessful()
protected void startSharedConnection()
Connection.start()
, relying on listeners to perform
appropriate recovery.startSharedConnection
in class AbstractJmsListeningContainer
Connection.start()
protected void stopSharedConnection()
Connection.stop()
, relying on listeners to perform
appropriate recovery after a restart.stopSharedConnection
in class AbstractJmsListeningContainer
Connection.start()
protected void handleListenerSetupFailure(Throwable ex, boolean alreadyRecovered)
The default implementation logs the exception at warn level if not recovered yet, and at debug level if already recovered. Can be overridden in subclasses.
ex
- the exception to handlealreadyRecovered
- whether a previously executing listener
already recovered from the present listener setup failure
(this usually indicates a follow-up failure than can be ignored
other than for debug log purposes)recoverAfterListenerSetupFailure()
protected void recoverAfterListenerSetupFailure()
The default implementation delegates to DefaultMessageListenerContainer's
recovery-capable refreshConnectionUntilSuccessful()
method, which will
try to re-establish a Connection to the JMS provider both for the shared
and the non-shared Connection case.
protected void refreshConnectionUntilSuccessful()
The default implementation retries until it successfully established a Connection, for as long as this message listener container is running. Applies the specified recovery interval between retries.
protected void refreshDestination()
Called after listener setup failure, assuming that a cached Destination object might have become invalid (a typical case on WebLogic JMS).
The default implementation removes the destination from a DestinationResolver's cache, in case of a CachingDestinationResolver.
protected void sleepInbetweenRecoveryAttempts()
public final boolean isRecovering()
May be used to detect recovery phases but also the end of a recovery phase,
with isRecovering()
switching to false
after having been found
to return true
before.
recoverAfterListenerSetupFailure()