Class DirectMessageListenerContainer

java.lang.Object

org.springframework.amqp.rabbit.connection.RabbitAccessor

org.springframework.amqp.rabbit.listener.ObservableListenerContainer

org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer

org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer

All Implemented Interfaces:

MessageListenerContainer, Aware, BeanNameAware, DisposableBean, InitializingBean, ApplicationContextAware, ApplicationEventPublisherAware, Lifecycle, Phased, SmartLifecycle

Direct Known Subclasses:

DirectReplyToMessageListenerContainer

public class DirectMessageListenerContainer
extends AbstractMessageListenerContainer

The SimpleMessageListenerContainer is not so simple. Recent changes to the rabbitmq java client has facilitated a much simpler listener container that invokes the listener directly on the rabbit client consumer thread. There is no txSize property - each message is acked (or nacked) individually.

Since:

2.0

Author:

Gary Russell, Artem Bilan, Nicolas Ristock, Cao Weibo

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
protected final class	DirectMessageListenerContainer.SimpleConsumer	The consumer object.

Nested classes/interfaces inherited from class org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer

AbstractMessageListenerContainer.DefaultExclusiveConsumerLogger, AbstractMessageListenerContainer.JavaLangErrorHandler, AbstractMessageListenerContainer.SharedConnectionNotInitializedException, AbstractMessageListenerContainer.WrappedTransactionException

Field Summary

Fields

Modifier and Type	Field	Description
protected final List<DirectMessageListenerContainer.SimpleConsumer>	consumers

Fields inherited from class org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer

consumersLock, DEFAULT_DEBATCHING_ENABLED, DEFAULT_PREFETCH_COUNT, DEFAULT_RECOVERY_INTERVAL, DEFAULT_SHUTDOWN_TIMEOUT, lifecycleLock, stopNow

Fields inherited from class org.springframework.amqp.rabbit.connection.RabbitAccessor

logger

Fields inherited from interface org.springframework.context.SmartLifecycle

DEFAULT_PHASE

Constructor Summary

Constructors

Constructor	Description
DirectMessageListenerContainer()	Create an instance; RabbitAccessor.setConnectionFactory(ConnectionFactory) must be called before starting.
DirectMessageListenerContainer(ConnectionFactory connectionFactory)	Create an instance with the provided connection factory.

Method Summary

Modifier and Type	Method	Description
protected void	actualStart()
void	addQueueNames(String... queueNames)	Add queue(s) to this container's list of queues.
void	addQueues(Queue... queues)	Add queue(s) to this container's list of queues.
protected void	checkConnect()
protected void	consumerRemoved(DirectMessageListenerContainer.SimpleConsumer consumer)	Called whenever a consumer is removed.
void	destroy()	Calls AbstractMessageListenerContainer.shutdown() when the BeanFactory destroys the container instance.
protected void	doInitialize()	Register any invokers within this container.
protected void	doRedeclareElementsIfNecessary()
protected void	doStart()	Start this container, and notify all invoker tasks.
protected void	doStop()	This method is invoked when the container is stopping.
protected int	findIdleConsumer()	When adjusting down, return a consumer that can be canceled.
protected void	processMonitorTask()	Subclasses can override this to take additional actions when the monitor task runs.
boolean	removeQueueNames(String... queueNames)	Remove queue(s) from this container's list of queues.
boolean	removeQueues(Queue... queues)	Remove queue(s) from this container's list of queues.
void	setAckTimeout(long ackTimeout)	An approximate timeout; when messagesPerAck is greater than 1, and this time elapses since the last ack, the pending acks will be sent either when the next message arrives, or a short time later if no additional messages arrive.
void	setConsumersPerQueue(int consumersPerQueue)	Each queue runs in its own consumer; set this property to create multiple consumers for each queue.
final void	setExclusive(boolean exclusive)	Set to true for an exclusive consumer - if true, the consumers per queue must be 1.
void	setMessagesPerAck(int messagesPerAck)	Set the number of messages to receive before acknowledging (success).
final void	setMissingQueuesFatal(boolean missingQueuesFatal)	If all the configured queue(s) are not available on the broker, this setting determines whether the condition is fatal.
void	setMonitorInterval(long monitorInterval)	Set how often to run a task to check for failed consumers and idle containers.
void	setQueueNames(String... queueName)	Set the name of the queue(s) to receive messages from.
void	setTaskScheduler(TaskScheduler taskScheduler)	Set the task scheduler to use for the task that monitors idle containers and failed consumers.
protected void	shutdownAndWaitOrCallback(Runnable callback)
void	stop(Runnable callback)

Methods inherited from class org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer

actualInvokeListener, addAfterReceivePostProcessors, afterPropertiesSet, causeChainHasImmediateAcknowledgeAmqpException, checkMessageListener, checkMismatchedQueues, configureAdminIfNeeded, debatch, doInvokeListener, doInvokeListener, doSetPossibleAuthenticationFailureFatal, doShutdown, executeListener, executeListenerAndHandleException, getAcknowledgeMode, getAdviceChain, getAfterReceivePostProcessors, getAmqpAdmin, getApplicationEventPublisher, getBatchingStrategy, getConnectionFactory, getConsumeDelay, getConsumerArguments, getConsumerTagStrategy, getErrorHandler, getExclusiveConsumerExceptionLogger, getFailedDeclarationRetryInterval, getIdleEventInterval, getJavaLangErrorHandler, getLastReceive, getMessageAckListener, getMessageListener, getMessagePropertiesConverter, getPhase, getPrefetchCount, getQueueNames, getQueueNamesAsSet, getQueueNamesToQueues, getRecoveryBackOff, getRoutingConnectionFactory, getRoutingLookupKey, getShutdownTimeout, getTaskExecutor, getTransactionAttribute, getTransactionManager, handleListenerException, initialize, initializeProxy, invokeErrorHandler, invokeListener, isActive, isAlwaysRequeueWithTxManagerRollback, isAsyncReplies, isAutoDeclare, isAutoStartup, isChannelLocallyTransacted, isDeBatchingEnabled, isDefaultRequeueRejected, isExclusive, isExposeListenerChannel, isForceCloseChannel, isForceStop, isGlobalQos, isMismatchedQueuesFatal, isMissingQueuesFatal, isMissingQueuesFatalSet, isNoLocal, isPossibleAuthenticationFailureFatal, isPossibleAuthenticationFailureFatalSet, isRunning, isStatefulRetryFatalWithNullMessageId, lazyLoad, prepareHolderForRollback, publishConsumerFailedEvent, publishIdleContainerEvent, publishMissingQueueEvent, redeclareElementsIfNecessary, removeAfterReceivePostProcessor, setAcknowledgeMode, setAdviceChain, setAfterReceivePostProcessors, setAlwaysRequeueWithTxManagerRollback, setAmqpAdmin, setApplicationEventPublisher, setAutoDeclare, setAutoStartup, setBatchingStrategy, setConsumeDelay, setConsumerArguments, setConsumerTagStrategy, setDeBatchingEnabled, setDefaultRequeueRejected, setErrorHandler, setErrorHandlerLoggerName, setExclusiveConsumerExceptionLogger, setExposeListenerChannel, setFailedDeclarationRetryInterval, setForceCloseChannel, setForceStop, setGlobalQos, setIdleEventInterval, setjavaLangErrorHandler, setLookupKeyQualifier, setMessageAckListener, setMessageListener, setMessagePropertiesConverter, setMismatchedQueuesFatal, setNoLocal, setNotRunning, setObservationConvention, setPhase, setPossibleAuthenticationFailureFatal, setPrefetchCount, setQueues, setRecoveryBackOff, setRecoveryInterval, setShutdownTimeout, setStatefulRetryFatalWithNullMessageId, setTaskExecutor, setTransactionAttribute, setTransactionManager, setupMessageListener, shutdown, shutdown, start, stop, updateLastReceive, validateConfiguration, wrapToListenerExecutionFailedExceptionIfNeeded

Methods inherited from class org.springframework.amqp.rabbit.listener.ObservableListenerContainer

checkMicrometer, checkObservation, getApplicationContext, getBeanName, getListenerId, getMicrometerHolder, isApplicationContextClosed, setApplicationContext, setBeanName, setListenerId, setMicrometerEnabled, setMicrometerTags, setObservationEnabled

Methods inherited from class org.springframework.amqp.rabbit.connection.RabbitAccessor

convertRabbitAccessException, createConnection, getChannel, getConnection, getObservationRegistry, getTransactionalResourceHolder, isChannelTransacted, obtainObservationRegistry, setChannelTransacted, setConnectionFactory

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.springframework.amqp.rabbit.listener.MessageListenerContainer

isConsumerBatchEnabled

Field Details

consumers

protected final List<DirectMessageListenerContainer.SimpleConsumer> consumers

Constructor Details

DirectMessageListenerContainer

public DirectMessageListenerContainer()

Create an instance; RabbitAccessor.setConnectionFactory(ConnectionFactory) must be called before starting.

DirectMessageListenerContainer

public DirectMessageListenerContainer(ConnectionFactory connectionFactory)

Create an instance with the provided connection factory.

Parameters:

connectionFactory - the connection factory.

Method Details

setConsumersPerQueue

public void setConsumersPerQueue(int consumersPerQueue)

Each queue runs in its own consumer; set this property to create multiple consumers for each queue. If the container is already running, the number of consumers per queue will be adjusted up or down as necessary.

Parameters:

consumersPerQueue - the consumers per queue.

setExclusive

public final void setExclusive(boolean exclusive)

Set to true for an exclusive consumer - if true, the consumers per queue must be 1.

Overrides:

setExclusive in class AbstractMessageListenerContainer

Parameters:

exclusive - true for an exclusive consumer.

setTaskScheduler

public void setTaskScheduler(TaskScheduler taskScheduler)

Set the task scheduler to use for the task that monitors idle containers and failed consumers.

Parameters:

taskScheduler - the scheduler.

setMonitorInterval

public void setMonitorInterval(long monitorInterval)

Set how often to run a task to check for failed consumers and idle containers.

Parameters:

monitorInterval - the interval; default 10000 but it will be adjusted down to the smallest of this, idleEventInterval / 2 (if configured) or failedDeclarationRetryInterval.

setQueueNames

public void setQueueNames(String... queueName)

Description copied from class: AbstractMessageListenerContainer

Set the name of the queue(s) to receive messages from.

Specified by:

setQueueNames in interface MessageListenerContainer

Overrides:

setQueueNames in class AbstractMessageListenerContainer

Parameters:

queueName - the desired queueName(s) (can not be null)

setMissingQueuesFatal

public final void setMissingQueuesFatal(boolean missingQueuesFatal)

If all the configured queue(s) are not available on the broker, this setting determines whether the condition is fatal. When true, and the queues are missing during startup, the context refresh() will fail.

When false, the condition is not considered fatal and the container will continue to attempt to start the consumers.

Defaults to false for this container.

Overrides:

setMissingQueuesFatal in class AbstractMessageListenerContainer

Parameters:

missingQueuesFatal - the missingQueuesFatal to set.

See Also:

• AbstractMessageListenerContainer.setAutoDeclare(boolean)

setMessagesPerAck

public void setMessagesPerAck(int messagesPerAck)

Set the number of messages to receive before acknowledging (success). A failed message will short-circuit this counter.

Parameters:

messagesPerAck - the number of messages.

See Also:

• setAckTimeout(long)

setAckTimeout

public void setAckTimeout(long ackTimeout)

An approximate timeout; when messagesPerAck is greater than 1, and this time elapses since the last ack, the pending acks will be sent either when the next message arrives, or a short time later if no additional messages arrive. In that case, the actual time depends on the monitorInterval.

Parameters:

ackTimeout - the timeout in milliseconds (default 20000);

See Also:

• setMessagesPerAck(int)

addQueueNames

public void addQueueNames(String... queueNames)

Description copied from class: AbstractMessageListenerContainer

Add queue(s) to this container's list of queues.

Overrides:

addQueueNames in class AbstractMessageListenerContainer

Parameters:

queueNames - The queue(s) to add.

addQueues

public void addQueues(Queue... queues)

Description copied from class: AbstractMessageListenerContainer

Add queue(s) to this container's list of queues.

Overrides:

addQueues in class AbstractMessageListenerContainer

Parameters:

queues - The queue(s) to add.

removeQueueNames

public boolean removeQueueNames(String... queueNames)

Description copied from class: AbstractMessageListenerContainer

Remove queue(s) from this container's list of queues.

Overrides:

removeQueueNames in class AbstractMessageListenerContainer

Parameters:

queueNames - The queue(s) to remove.

Returns:

the boolean result of removal on the target queueNames List.

removeQueues

public boolean removeQueues(Queue... queues)

Description copied from class: AbstractMessageListenerContainer

Remove queue(s) from this container's list of queues.

Overrides:

removeQueues in class AbstractMessageListenerContainer

Parameters:

queues - The queue(s) to remove.

Returns:

the boolean result of removal on the target queueNames List.

findIdleConsumer

protected int findIdleConsumer()

When adjusting down, return a consumer that can be canceled. Called while locked on AbstractMessageListenerContainer.consumersLock.

Returns:

the consumer index or -1 if non-idle.

Since:

2.0.6

doInitialize

protected void doInitialize()

Description copied from class: AbstractMessageListenerContainer

Register any invokers within this container.

Subclasses need to implement this method for their specific invoker management process.

Specified by:

doInitialize in class AbstractMessageListenerContainer

doStart

protected void doStart()

Description copied from class: AbstractMessageListenerContainer

Start this container, and notify all invoker tasks.

Overrides:

doStart in class AbstractMessageListenerContainer

stop

public void stop(Runnable callback)

Specified by:

stop in interface SmartLifecycle

Overrides:

stop in class AbstractMessageListenerContainer

doStop

protected void doStop()

Description copied from class: AbstractMessageListenerContainer

This method is invoked when the container is stopping.

Overrides:

doStop in class AbstractMessageListenerContainer

actualStart

protected void actualStart()

checkConnect

protected void checkConnect()

doRedeclareElementsIfNecessary

protected void doRedeclareElementsIfNecessary()

processMonitorTask

protected void processMonitorTask()

Subclasses can override this to take additional actions when the monitor task runs.

shutdownAndWaitOrCallback

protected void shutdownAndWaitOrCallback(@Nullable
 Runnable callback)

Overrides:

shutdownAndWaitOrCallback in class AbstractMessageListenerContainer

consumerRemoved

protected void consumerRemoved(DirectMessageListenerContainer.SimpleConsumer consumer)

Called whenever a consumer is removed.

Parameters:

consumer - the consumer.

destroy

public void destroy()

Description copied from class: AbstractMessageListenerContainer

Calls AbstractMessageListenerContainer.shutdown() when the BeanFactory destroys the container instance.

Specified by:

destroy in interface DisposableBean

Overrides:

destroy in class AbstractMessageListenerContainer

See Also:

• AbstractMessageListenerContainer.shutdown()