org.springframework.amqp.rabbit.connection

Class CachingConnectionFactory

java.lang.Object

org.springframework.amqp.rabbit.connection.AbstractConnectionFactory

org.springframework.amqp.rabbit.connection.CachingConnectionFactory

All Implemented Interfaces:

com.rabbitmq.client.ShutdownListener, EventListener, ConnectionFactory, Aware, BeanNameAware, DisposableBean, InitializingBean, ApplicationContextAware, ApplicationEventPublisherAware, ApplicationListener<ContextClosedEvent>

@ManagedResource
public class CachingConnectionFactory
extends AbstractConnectionFactory
implements InitializingBean, com.rabbitmq.client.ShutdownListener

A ConnectionFactory implementation that (when the cache mode is CachingConnectionFactory.CacheMode.CHANNEL (default) returns the same Connection from all createConnection() calls, and ignores calls to Connection.close() and caches Channel.

By default, only one Channel will be cached, with further requested Channels being created and disposed on demand. Consider raising the "channelCacheSize" value in case of a high-concurrency environment.

When the cache mode is CachingConnectionFactory.CacheMode.CONNECTION, a new (or cached) connection is used for each createConnection(); connections are cached according to the "connectionCacheSize" value. Both connections and channels are cached in this mode.

CachingConnectionFactory.CacheMode.CONNECTION is not compatible with a Rabbit Admin that auto-declares queues etc.

NOTE: This ConnectionFactory requires explicit closing of all Channels obtained form its Connection(s). This is the usual recommendation for native Rabbit access code anyway. However, with this ConnectionFactory, its use is mandatory in order to actually allow for Channel reuse. Channel.close() returns the channel to the cache, if there is room, or physically closes the channel otherwise.

Author:

Mark Pollack, Mark Fisher, Dave Syer, Gary Russell, Artem Bilan, Steve Powell, Will Droste

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	CachingConnectionFactory.CacheMode The cache mode.
static class	CachingConnectionFactory.ConfirmType The type of publisher confirms to use.

Nested classes/interfaces inherited from class org.springframework.amqp.rabbit.connection.AbstractConnectionFactory

AbstractConnectionFactory.AddressShuffleMode

Field Summary

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

DEFAULT_CLOSE_TIMEOUT, logger

Constructor Summary

Constructors

Constructor and Description
CachingConnectionFactory() Create a new CachingConnectionFactory initializing the hostname to be the value returned from InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws an exception.
CachingConnectionFactory(com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory) Create a new CachingConnectionFactory for the given target ConnectionFactory.
CachingConnectionFactory(int port) Create a new CachingConnectionFactory given a port on the hostname returned from InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws an exception.
CachingConnectionFactory(String hostname) Create a new CachingConnectionFactory given a host name.
CachingConnectionFactory(String hostNameArg, int port) Create a new CachingConnectionFactory given a host name and port.
CachingConnectionFactory(URI uri) Create a new CachingConnectionFactory given a URI.

Method Summary

Modifier and Type	Method and Description
void	addConnectionListener(ConnectionListener listener)
void	afterPropertiesSet()
protected void	closeAndClear(Collection<ChannelProxy> theChannels)
protected void	closeChannels(Collection<ChannelProxy> theChannels)
Connection	createConnection()
void	destroy() Close the underlying shared connection.
CachingConnectionFactory.CacheMode	getCacheMode()
Properties	getCacheProperties()
int	getChannelCacheSize()
protected ExecutorService	getChannelsExecutor() Determine the executor service used for target channels.
int	getConnectionCacheSize()
Properties	getPublisherConnectionFactoryCacheProperties() Return the cache properties from the underlying publisher sub-factory.
boolean	isPublisherConfirms() Return true if publisher confirms are enabled.
boolean	isPublisherReturns() Return true if publisher returns are enabled.
boolean	isSimplePublisherConfirms() Return true if simple publisher confirms are enabled.
protected void	reset(List<ChannelProxy> channels, List<ChannelProxy> txChannels, Map<com.rabbitmq.client.Channel,ChannelProxy> channelsAwaitingAcks)
void	resetConnection() Close the connection(s).
void	setCacheMode(CachingConnectionFactory.CacheMode cacheMode)
void	setChannelCacheSize(int sessionCacheSize) The number of channels to maintain in the cache.
void	setChannelCheckoutTimeout(long channelCheckoutTimeout) Sets the channel checkout timeout.
void	setConnectionCacheSize(int connectionCacheSize)
void	setConnectionLimit(int connectionLimit) Set the connection limit when using cache mode CONNECTION.
void	setConnectionListeners(List<? extends ConnectionListener> listeners)
void	setPublisherChannelFactory(PublisherCallbackChannelFactory publisherChannelFactory) Set the factory to use to create PublisherCallbackChannel instances.
void	setPublisherConfirms(boolean publisherConfirms) Deprecated. in favor of setPublisherConfirmType(ConfirmType).
void	setPublisherConfirmType(CachingConnectionFactory.ConfirmType confirmType) Set the confirm type to use; default CachingConnectionFactory.ConfirmType.NONE.
void	setPublisherConnectionFactory(AbstractConnectionFactory publisherConnectionFactory) Set a custom publisher connection factory; the type does not need to be the same as this factory.
void	setPublisherReturns(boolean publisherReturns)
void	setSimplePublisherConfirms(boolean simplePublisherConfirms) Deprecated. in favor of setPublisherConfirmType(ConfirmType).
String	toString()

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

addChannelListener, clearConnectionListeners, createBareConnection, doSetPublisherConnectionFactory, getApplicationContext, getApplicationEventPublisher, getBeanName, getChannelListener, getCloseTimeout, getConnectionListener, getConnectionNameStrategy, getContextStopped, getDefaultHostName, getExecutorService, getHost, getPort, getPublisherConnectionFactory, getRabbitConnectionFactory, getUsername, getVirtualHost, hasPublisherConnectionFactory, onApplicationEvent, removeConnectionListener, setAddresses, setAddressResolver, setAddressShuffleMode, setApplicationContext, setApplicationEventPublisher, setBeanName, setChannelListeners, setCloseExceptionLogger, setCloseTimeout, setConnectionNameStrategy, setConnectionThreadFactory, setConnectionTimeout, setExecutor, setHost, setPassword, setPort, setRecoveryListener, setRequestedHeartBeat, setShuffleAddresses, setUri, setUri, setUsername, setVirtualHost, shutdownCompleted

Methods inherited from class java.lang.Object

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

Methods inherited from interface com.rabbitmq.client.ShutdownListener

shutdownCompleted

Methods inherited from interface org.springframework.context.ApplicationListener

forPayload

Constructor Detail

CachingConnectionFactory

public CachingConnectionFactory()

Create a new CachingConnectionFactory initializing the hostname to be the value returned from InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws an exception.

CachingConnectionFactory

public CachingConnectionFactory(@Nullable
                                String hostname)

Create a new CachingConnectionFactory given a host name.

Parameters:

hostname - the host name to connect to

CachingConnectionFactory

public CachingConnectionFactory(int port)

Create a new CachingConnectionFactory given a port on the hostname returned from InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws an exception.

Parameters:

port - the port number

CachingConnectionFactory

public CachingConnectionFactory(@Nullable
                                String hostNameArg,
                                int port)

Create a new CachingConnectionFactory given a host name and port.

Parameters:

hostNameArg - the host name to connect to

port - the port number

CachingConnectionFactory

public CachingConnectionFactory(URI uri)

Create a new CachingConnectionFactory given a URI.

Parameters:

uri - the amqp uri configuring the connection

Since:

1.5

CachingConnectionFactory

public CachingConnectionFactory(com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory)

Create a new CachingConnectionFactory for the given target ConnectionFactory.

Parameters:

rabbitConnectionFactory - the target ConnectionFactory

Method Detail

setPublisherConnectionFactory

public void setPublisherConnectionFactory(@Nullable
                                          AbstractConnectionFactory publisherConnectionFactory)

Description copied from class: AbstractConnectionFactory

Set a custom publisher connection factory; the type does not need to be the same as this factory.

Overrides:

setPublisherConnectionFactory in class AbstractConnectionFactory

Parameters:

publisherConnectionFactory - the factory.

setChannelCacheSize

public void setChannelCacheSize(int sessionCacheSize)

The number of channels to maintain in the cache. By default, channels are allocated on demand (unbounded) and this represents the maximum cache size. To limit the available channels, see setChannelCheckoutTimeout(long).

Parameters:

sessionCacheSize - the channel cache size.

See Also:

setChannelCheckoutTimeout(long)

getChannelCacheSize

public int getChannelCacheSize()

getCacheMode

public CachingConnectionFactory.CacheMode getCacheMode()

setCacheMode

public void setCacheMode(CachingConnectionFactory.CacheMode cacheMode)

getConnectionCacheSize

public int getConnectionCacheSize()

setConnectionCacheSize

public void setConnectionCacheSize(int connectionCacheSize)

setConnectionLimit

public void setConnectionLimit(int connectionLimit)

Set the connection limit when using cache mode CONNECTION. When the limit is reached and there are no idle connections, the channelCheckoutTimeLimit is used to wait for a connection to become idle.

Parameters:

connectionLimit - the limit.

Since:

1.5.5

isPublisherConfirms

public boolean isPublisherConfirms()

Description copied from interface: ConnectionFactory

Return true if publisher confirms are enabled.

Specified by:

isPublisherConfirms in interface ConnectionFactory

Returns:

publisherConfirms.

isPublisherReturns

public boolean isPublisherReturns()

Description copied from interface: ConnectionFactory

Return true if publisher returns are enabled.

Specified by:

isPublisherReturns in interface ConnectionFactory

Returns:

publisherReturns.

setPublisherReturns

public void setPublisherReturns(boolean publisherReturns)

setPublisherConfirms

@Deprecated
public void setPublisherConfirms(boolean publisherConfirms)

Deprecated. in favor of setPublisherConfirmType(ConfirmType).

Use full (correlated) publisher confirms, with correlation data and a callback for each message.

Parameters:

publisherConfirms - true for full publisher returns,

Since:

1.1

See Also:

setSimplePublisherConfirms(boolean)

setSimplePublisherConfirms

@Deprecated
public void setSimplePublisherConfirms(boolean simplePublisherConfirms)

Deprecated. in favor of setPublisherConfirmType(ConfirmType).

Use simple publisher confirms where the template simply waits for completion.

Parameters:

simplePublisherConfirms - true for confirms.

Since:

2.1

See Also:

setPublisherConfirms(boolean)

isSimplePublisherConfirms

public boolean isSimplePublisherConfirms()

Description copied from interface: ConnectionFactory

Return true if simple publisher confirms are enabled.

Specified by:

isSimplePublisherConfirms in interface ConnectionFactory

Returns:

simplePublisherConfirms

setPublisherConfirmType

public void setPublisherConfirmType(CachingConnectionFactory.ConfirmType confirmType)

Set the confirm type to use; default CachingConnectionFactory.ConfirmType.NONE.

Parameters:

confirmType - the confirm type.

Since:

2.2

setChannelCheckoutTimeout

public void setChannelCheckoutTimeout(long channelCheckoutTimeout)

Sets the channel checkout timeout. When greater than 0, enables channel limiting in that the channelCacheSize becomes the total number of available channels per connection rather than a simple cache size. Note that changing the channelCacheSize does not affect the limit on existing connection(s), invoke destroy() to cause a new connection to be created with the new limit.

Since 1.5.5, also applies to getting a connection when the cache mode is CONNECTION.

Parameters:

channelCheckoutTimeout - the timeout in milliseconds; default 0 (channel limiting not enabled).

Since:

1.4.2

See Also:

setConnectionLimit(int)

setPublisherChannelFactory

public void setPublisherChannelFactory(PublisherCallbackChannelFactory publisherChannelFactory)

Set the factory to use to create PublisherCallbackChannel instances.

Parameters:

publisherChannelFactory - the factory.

Since:

2.1.6

afterPropertiesSet

public void afterPropertiesSet()

Specified by:

afterPropertiesSet in interface InitializingBean

setConnectionListeners

public void setConnectionListeners(List<? extends ConnectionListener> listeners)

Overrides:

setConnectionListeners in class AbstractConnectionFactory

addConnectionListener

public void addConnectionListener(ConnectionListener listener)

Specified by:

addConnectionListener in interface ConnectionFactory

Overrides:

addConnectionListener in class AbstractConnectionFactory

createConnection

public final Connection createConnection()
                                  throws AmqpException

Specified by:

createConnection in interface ConnectionFactory

Throws:

AmqpException

destroy

public final void destroy()

Close the underlying shared connection. Use resetConnection() to close the connection while the application is still running.

As this bean implements DisposableBean, a bean factory will automatically invoke this on destruction of its cached singletons.

If called after the context is closed, the connection factory can no longer server up connections.

Specified by:

destroy in interface DisposableBean

Overrides:

destroy in class AbstractConnectionFactory

resetConnection

public void resetConnection()

Close the connection(s). This will impact any in-process operations. New connection(s) will be created on demand after this method returns. This might be used to force a reconnect to the primary broker after failing over to a secondary broker.

reset

protected void reset(List<ChannelProxy> channels,
                     List<ChannelProxy> txChannels,
                     Map<com.rabbitmq.client.Channel,ChannelProxy> channelsAwaitingAcks)

closeAndClear

protected void closeAndClear(Collection<ChannelProxy> theChannels)

closeChannels

protected void closeChannels(Collection<ChannelProxy> theChannels)

getCacheProperties

@ManagedAttribute
public Properties getCacheProperties()

getPublisherConnectionFactoryCacheProperties

@ManagedAttribute
public Properties getPublisherConnectionFactoryCacheProperties()

Return the cache properties from the underlying publisher sub-factory.

Returns:

the properties.

Since:

2.0.2

getChannelsExecutor

protected ExecutorService getChannelsExecutor()

Determine the executor service used for target channels.

Returns:

specified executor service otherwise the default one is created and returned.

Since:

1.7.9

toString

public String toString()

Overrides:

toString in class AbstractConnectionFactory