程序包 org.springframework.jms.listener

类 DefaultMessageListenerContainer

  • 所有已实现的接口:
    Aware, BeanNameAware, DisposableBean, InitializingBean, Lifecycle, Phased, SmartLifecycle, MessageListenerContainer
    public class DefaultMessageListenerContainer
extends AbstractPollingMessageListenerContainer
    Message listener container variant that uses plain JMS client APIs, specifically a loop of 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 Java EE 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 Java EE 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 Java EE environment, in combination with a JTA-aware JMS ConnectionFactory obtained from JNDI (check your Java EE 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 Java EE 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.

    从以下版本开始:
    2.0
    作者:
    Juergen Hoeller
    另请参阅:
    AbstractPollingMessageListenerContainer.setTransactionManager(org.springframework.transaction.PlatformTransactionManager), setCacheLevel(int), MessageConsumer.receive(long), SimpleMessageListenerContainer, JmsMessageEndpointManager

    • 方法详细资料

      • setTaskExecutor

        public void setTaskExecutor​(Executor taskExecutor)
        Set the Spring 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 Java EE environment. A plain thread pool does not add much value, as this listener container will occupy a number of threads for its entire lifetime.

        另请参阅:
        setConcurrentConsumers(int), SimpleAsyncTaskExecutor, WorkManagerTaskExecutor

      • getCacheLevel

        public int getCacheLevel()
        Return the level of caching that this listener container is allowed to apply.

      • setConcurrentConsumers

        public void setConcurrentConsumers​(int concurrentConsumers)
        Specify the number of concurrent consumers to create. Default is 1.

        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)

      • setMaxConcurrentConsumers

        public void setMaxConcurrentConsumers​(int maxConcurrentConsumers)
        Specify the maximum number of concurrent consumers to create. Default is 1.

        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)

      • 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.

        另请参阅:
        setTaskExecutor(java.util.concurrent.Executor), AbstractPollingMessageListenerContainer.setReceiveTimeout(long), SchedulingTaskExecutor.prefersShortLivedTasks()

      • getMaxMessagesPerTask

        public final int getMaxMessagesPerTask()
        Return the maximum number of messages to process in one task.

      • setIdleConsumerLimit

        public void setIdleConsumerLimit​(int idleConsumerLimit)
        Specify the limit for the number of consumers that are allowed to be idle at any given time.

        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.

      • getIdleConsumerLimit

        public final int getIdleConsumerLimit()
        Return the limit for the number of idle consumers.

      • setIdleTaskExecutionLimit

        public void setIdleTaskExecutionLimit​(int idleTaskExecutionLimit)
        Specify the limit for idle executions of a consumer 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.

        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.

        另请参阅:
        setMaxMessagesPerTask(int), AbstractPollingMessageListenerContainer.setReceiveTimeout(long)

      • getIdleTaskExecutionLimit

        public final int getIdleTaskExecutionLimit()
        Return the limit for idle executions of a consumer task.

      • stop

        public void stop​(Runnable callback)
          throws JmsException
        Stop this listener container, invoking the specific callback once all listener processing has actually stopped.

        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.

        参数:
        callback - the callback to invoke once listener processing has fully stopped
        抛出:
        JmsException - if stopping failed
        另请参阅:
        AbstractJmsListeningContainer.stop()

      • isRegisteredWithDestination

        public boolean isRegisteredWithDestination()
        Return whether at least one consumer has entered a fixed registration with the target destination. This is particularly interesting for the pub-sub case where it might be important to have an actual consumer registered that is guaranteed not to miss any messages that are just about to be published.

        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.

      • handleListenerSetupFailure

        protected void handleListenerSetupFailure​(Throwable ex,
                                          boolean alreadyRecovered)
        Handle the given exception that arose during setup of a listener. Called for every such exception in every concurrent listener.

        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 handle
        alreadyRecovered - 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()

      • refreshConnectionUntilSuccessful

        protected void refreshConnectionUntilSuccessful()
        Refresh the underlying Connection, not returning before an attempt has been successful. Called in case of a shared Connection as well as without shared Connection, so either needs to operate on the shared Connection or on a temporary Connection that just gets established for validation purposes.

        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.

        另请参阅:
        setRecoveryInterval(long), start(), AbstractJmsListeningContainer.stop()

      • applyBackOffTime

        protected boolean applyBackOffTime​(BackOffExecution execution)
        Apply the next back-off time using the specified BackOffExecution.

        Return true if the back-off period has been applied and a new attempt to recover should be made, false if no further attempt should be made.

        从以下版本开始:
        4.1

      • isRecovering

        public final boolean isRecovering()
        Return whether this listener container is currently in a recovery attempt.

        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()