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, PublisherCallbackChannelConnectionFactory, org.springframework.beans.factory.Aware, org.springframework.beans.factory.BeanNameAware, org.springframework.beans.factory.DisposableBean, org.springframework.beans.factory.InitializingBean, org.springframework.context.ApplicationContextAware, org.springframework.context.ApplicationListener<org.springframework.context.event.ContextClosedEvent>, org.springframework.context.Lifecycle, org.springframework.context.Phased, org.springframework.context.SmartLifecycle

public class CachingConnectionFactory
extends AbstractConnectionFactory
implements org.springframework.beans.factory.InitializingBean, com.rabbitmq.client.ShutdownListener, org.springframework.context.ApplicationContextAware, org.springframework.context.ApplicationListener<org.springframework.context.event.ContextClosedEvent>, PublisherCallbackChannelConnectionFactory, org.springframework.context.SmartLifecycle

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

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	CachingConnectionFactory.CacheMode

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 hostname, 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()
Connection	createConnection()
void	destroy() Close the underlying shared connection.
CachingConnectionFactory.CacheMode	getCacheMode()
int	getChannelCacheSize()
int	getConnectionCachesize() Deprecated.
int	getConnectionCacheSize()
int	getPhase()
boolean	isAutoStartup()
boolean	isPublisherConfirms()
boolean	isPublisherReturns()
boolean	isRunning()
void	onApplicationEvent(org.springframework.context.event.ContextClosedEvent event)
protected void	reset(List<ChannelProxy> channels, List<ChannelProxy> txChannels)
void	resetConnection() Close the connection(s).
void	setApplicationContext(org.springframework.context.ApplicationContext applicationContext)
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	setCloseExceptionLogger(ConditionalExceptionLogger closeExceptionLogger) Set the strategy for logging close exceptions; by default, if a channel is closed due to a failed passive queue declaration, it is logged at debug level.
void	setConnectionCacheSize(int connectionCacheSize)
void	setConnectionLimit(int connectionLimit) Set the connection limit when using cache mode CONNECTION.
void	setConnectionListeners(List<? extends ConnectionListener> listeners)
void	setPhase(int phase) Defaults to phase - 1000 so the factory is stopped in a very late phase, allowing other beans to use the connection to clean up.
void	setPublisherConfirms(boolean publisherConfirms)
void	setPublisherReturns(boolean publisherReturns)
void	shutdownCompleted(com.rabbitmq.client.ShutdownSignalException cause)
void	start()
void	stop() Stop the connection factory to prevent its connection from being used.
void	stop(Runnable callback)
String	toString()

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

addChannelListener, clearConnectionListeners, createBareConnection, getChannelListener, getCloseTimeout, getConnectionListener, getDefaultHostName, getExecutorService, getHost, getPort, getRabbitConnectionFactory, getVirtualHost, removeConnectionListener, setAddresses, setBeanName, setChannelListeners, setCloseTimeout, setConnectionThreadFactory, setConnectionTimeout, setExecutor, setHost, setPassword, setPort, setRequestedHeartBeat, setUri, setUri, setUsername, setVirtualHost

Methods inherited from class java.lang.Object

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

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(String hostname,
                                int port)

Create a new CachingConnectionFactory given a host name and port.

Parameters:

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

Create a new CachingConnectionFactory given a host name.

Parameters:

hostname - the host name to connect to

CachingConnectionFactory

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

Create a new CachingConnectionFactory for the given target ConnectionFactory.

Parameters:

rabbitConnectionFactory - the target ConnectionFactory

Method Detail

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

@Deprecated
public int getConnectionCachesize()

Deprecated.

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

Specified by:

isPublisherConfirms in interface PublisherCallbackChannelConnectionFactory

Returns:

true if publisher confirms are enabled.

isPublisherReturns

public boolean isPublisherReturns()

Specified by:

isPublisherReturns in interface PublisherCallbackChannelConnectionFactory

Returns:

true if publisher returns are enabled.

setPublisherReturns

public void setPublisherReturns(boolean publisherReturns)

setPublisherConfirms

public void setPublisherConfirms(boolean publisherConfirms)

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)

setCloseExceptionLogger

public void setCloseExceptionLogger(ConditionalExceptionLogger closeExceptionLogger)

Set the strategy for logging close exceptions; by default, if a channel is closed due to a failed passive queue declaration, it is logged at debug level. Normal channel closes (200 OK) are not logged. All others are logged at ERROR level (unless access is refused due to an exclusive consumer condition, in which case, it is logged at INFO level).

Parameters:

closeExceptionLogger - the ConditionalExceptionLogger.

Since:

1.5

afterPropertiesSet

public void afterPropertiesSet()
                        throws Exception

Specified by:

afterPropertiesSet in interface org.springframework.beans.factory.InitializingBean

Throws:

Exception

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

shutdownCompleted

public void shutdownCompleted(com.rabbitmq.client.ShutdownSignalException cause)

Specified by:

shutdownCompleted in interface com.rabbitmq.client.ShutdownListener

setApplicationContext

public void setApplicationContext(org.springframework.context.ApplicationContext applicationContext)
                           throws org.springframework.beans.BeansException

Specified by:

setApplicationContext in interface org.springframework.context.ApplicationContextAware

Throws:

org.springframework.beans.BeansException

onApplicationEvent

public void onApplicationEvent(org.springframework.context.event.ContextClosedEvent event)

Specified by:

onApplicationEvent in interface org.springframework.context.ApplicationListener<org.springframework.context.event.ContextClosedEvent>

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. The provider of this ConnectionFactory needs to care for proper shutdown.

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

Specified by:

destroy in interface org.springframework.beans.factory.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.

start

public void start()

Specified by:

start in interface org.springframework.context.Lifecycle

isRunning

public boolean isRunning()

Specified by:

isRunning in interface org.springframework.context.Lifecycle

getPhase

public int getPhase()

Specified by:

getPhase in interface org.springframework.context.Phased

setPhase

public void setPhase(int phase)

Defaults to phase - 1000 so the factory is stopped in a very late phase, allowing other beans to use the connection to clean up.

Parameters:

phase - the phase.

Since:

1.5.3

See Also:

getPhase()

isAutoStartup

public boolean isAutoStartup()

Specified by:

isAutoStartup in interface org.springframework.context.SmartLifecycle

stop

public void stop()

Stop the connection factory to prevent its connection from being used. Note: ignored unless the application context is in the process of being stopped.

Specified by:

stop in interface org.springframework.context.Lifecycle

stop

public void stop(Runnable callback)

Specified by:

stop in interface org.springframework.context.SmartLifecycle

reset

protected void reset(List<ChannelProxy> channels,
                     List<ChannelProxy> txChannels)

toString

public String toString()

Overrides:

toString in class AbstractConnectionFactory