AMQP-608: Add new connection factories

JIRA https://jira.spring.io/browse/AMQP-608

Initial Commit

* Doc and Javadoc polishing.

build.gradle

@@ -41,6 +41,7 @@ ext {
 	assertjVersion = '3.15.0'
 	assertkVersion = '0.20'
 	commonsHttpClientVersion = '4.5.10'
+	commonsPoolVersion = '2.8.0'
 	googleJsr305Version = '3.0.2'
 	hamcrestVersion = '2.2'
 	jacksonVersion = '2.10.3'
@@ -334,6 +335,7 @@ project('spring-rabbit') {
 			exclude group: 'org.springframework'
 		}
 		optionalApi "com.jayway.jsonpath:json-path:$jaywayJsonPathVersion"
+		optionalApi "org.apache.commons:commons-pool2:$commonsPoolVersion"
 
 		testApi project(':spring-rabbit-junit')
 		testImplementation("com.willowtreeapps.assertk:assertk-jvm:$assertkVersion")

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/Connection.java

@@ -88,4 +88,10 @@ public interface Connection extends AutoCloseable {
 		return null;
 	}
 
+	/**
+	 * Close any channel associated with the current thread.
+	 */
+	default void closeThreadChannel() {
+	}
+
 }

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PooledChannelConnectionFactory.java

@@ -0,0 +1,246 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbit.connection;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiConsumer;
+
+import org.aopalliance.aop.Advice;
+import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+import org.apache.commons.logging.Log;
+import org.apache.commons.pool2.ObjectPool;
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.PooledObjectFactory;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
+import org.apache.commons.pool2.impl.GenericObjectPool;
+
+import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.aop.support.NameMatchMethodPointcutAdvisor;
+import org.springframework.util.Assert;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.ConnectionFactory;
+
+/**
+ * A very simple connection factory that caches channels using Apache Pool2
+ * {@link GenericObjectPool}s (one for transactional and one for non-transactional
+ * channels). The pools have default configuration but they can be configured using
+ * a callback.
+ *
+ * @author Gary Russell
+ * @since 2.3
+ *
+ */
+public class PooledChannelConnectionFactory extends AbstractConnectionFactory {
+
+	private volatile ConnectionWrapper connection;
+
+	private boolean simplePublisherConfirms;
+
+	private BiConsumer<GenericObjectPool<Channel>, Boolean> poolConfigurer = (pool, tx) -> { };
+
+	/**
+	 * Construct an instance.
+	 *
+	 * @param rabbitConnectionFactory the rabbitmq connection factory.
+	 */
+	public PooledChannelConnectionFactory(ConnectionFactory rabbitConnectionFactory) {
+		this(rabbitConnectionFactory, false);
+	}
+
+	/**
+	 * Construct an instance.
+	 *
+	 * @param rabbitConnectionFactory the rabbitmq connection factory.
+	 * @param isPublisher true if we are creating a publisher connection factory.
+	 */
+	private PooledChannelConnectionFactory(ConnectionFactory rabbitConnectionFactory, boolean isPublisher) {
+		super(rabbitConnectionFactory);
+		if (!isPublisher) {
+			setPublisherConnectionFactory(new PooledChannelConnectionFactory(rabbitConnectionFactory, true));
+		}
+	}
+
+	/**
+	 * Add a consumer to configure the object pool. The second argument is true when
+	 * called with the transactional pool.
+	 * @param poolConfigurer the configurer.
+	 */
+	public void setPoolConfigurer(BiConsumer<GenericObjectPool<Channel>, Boolean> poolConfigurer) {
+		Assert.notNull(poolConfigurer, "'poolConfigurer' cannot be null");
+		this.poolConfigurer = poolConfigurer;
+	}
+
+	@Override
+	public boolean isSimplePublisherConfirms() {
+		return this.simplePublisherConfirms;
+	}
+
+	/**
+	 * Enable simple publisher confirms.
+	 * @param simplePublisherConfirms true to enable.
+	 */
+	public void setSimplePublisherConfirms(boolean simplePublisherConfirms) {
+		this.simplePublisherConfirms = simplePublisherConfirms;
+	}
+
+	@Override
+	public synchronized Connection createConnection() throws AmqpException {
+		if (this.connection == null || !this.connection.isOpen()) {
+			Connection bareConnection = createBareConnection();
+			this.connection = new ConnectionWrapper(bareConnection.getDelegate(), getCloseTimeout(),
+					this.simplePublisherConfirms, this.logger, this.poolConfigurer);
+		}
+		return this.connection;
+	}
+
+	@Override
+	public synchronized void destroy() {
+		super.destroy();
+		if (this.connection != null) {
+			this.connection.forceClose();
+			this.connection = null;
+		}
+	}
+
+	private final static class ConnectionWrapper extends SimpleConnection {
+
+		private final Log logger;
+
+		private final ObjectPool<Channel> channels;
+
+		private final ObjectPool<Channel> txChannels;
+
+		private final boolean simplePublisherConfirms;
+
+		ConnectionWrapper(com.rabbitmq.client.Connection delegate, int closeTimeout, boolean simplePublisherConfirms,
+				Log logger, BiConsumer<GenericObjectPool<Channel>, Boolean> configurer) {
+
+			super(delegate, closeTimeout);
+			GenericObjectPool<Channel> pool = new GenericObjectPool<>(new ChannelFactory());
+			configurer.accept(pool, false);
+			this.channels = pool;
+			pool = new GenericObjectPool<>(new TxChannelFactory());
+			configurer.accept(pool, true);
+			this.txChannels = pool;
+			this.simplePublisherConfirms = simplePublisherConfirms;
+			this.logger = logger;
+		}
+
+		@Override
+		public Channel createChannel(boolean transactional) {
+			try {
+				return transactional ? this.txChannels.borrowObject() : this.channels.borrowObject();
+			}
+			catch (Exception e) {
+				throw RabbitExceptionTranslator.convertRabbitAccessException(e);
+			}
+		}
+
+		private Channel createProxy(Channel channel, boolean transacted) {
+			ProxyFactory pf = new ProxyFactory(channel);
+			AtomicReference<Channel> proxy = new AtomicReference<>();
+			Advice advice = new MethodInterceptor() {
+
+				@Override
+				public Object invoke(MethodInvocation invocation) throws Throwable {
+					if (transacted) {
+						ConnectionWrapper.this.txChannels.returnObject(proxy.get());
+					}
+					else {
+						ConnectionWrapper.this.channels.returnObject(proxy.get());
+					}
+					return null;
+				}
+
+			};
+			NameMatchMethodPointcutAdvisor advisor = new NameMatchMethodPointcutAdvisor(advice);
+			advisor.addMethodName("close");
+			pf.addAdvisor(advisor);
+			proxy.set((Channel) pf.getProxy());
+			return proxy.get();
+		}
+
+		@Override
+		public void close() {
+		}
+
+		void forceClose() {
+			super.close();
+			this.channels.close();
+			this.txChannels.close();
+		}
+
+		private class ChannelFactory implements PooledObjectFactory<Channel> {
+
+			@Override
+			public PooledObject<Channel> makeObject() throws Exception {
+				Channel channel = ConnectionWrapper.super.createChannel(false);
+				if (ConnectionWrapper.this.simplePublisherConfirms) {
+					try {
+						channel.confirmSelect();
+					}
+					catch (IOException e) {
+						throw RabbitExceptionTranslator.convertRabbitAccessException(e);
+					}
+				}
+				return new DefaultPooledObject<>(createProxy(channel, false));
+			}
+
+			@Override
+			public void destroyObject(PooledObject<Channel> p) throws Exception {
+				p.getObject().close();
+			}
+
+			@Override
+			public boolean validateObject(PooledObject<Channel> p) {
+				return p.getObject().isOpen();
+			}
+
+			@Override
+			public void activateObject(PooledObject<Channel> p) {
+			}
+
+			@Override
+			public void passivateObject(PooledObject<Channel> p) {
+			}
+
+		}
+
+		private final class TxChannelFactory extends ChannelFactory {
+
+			@Override
+			public PooledObject<Channel> makeObject() throws Exception {
+				Channel channel = ConnectionWrapper.super.createChannel(true);
+				try {
+					channel.txSelect();
+				}
+				catch (IOException e) {
+					throw RabbitExceptionTranslator.convertRabbitAccessException(e);
+				}
+				return new DefaultPooledObject<>(createProxy(channel, true));
+			}
+
+		}
+
+	}
+
+}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitAccessor.java

@@ -58,7 +58,7 @@ public abstract class RabbitAccessor implements InitializingBean {
 	 *
 	 * @param connectionFactory The connection factory.
 	 */
-	public final void setConnectionFactory(ConnectionFactory connectionFactory) {
+	public void setConnectionFactory(ConnectionFactory connectionFactory) {
 		this.connectionFactory = connectionFactory;
 	}
 

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ThreadChannelConnectionFactory.java

@@ -0,0 +1,209 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbit.connection;
+
+import java.io.IOException;
+import java.util.concurrent.TimeoutException;
+
+import org.aopalliance.aop.Advice;
+import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+
+import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.aop.support.NameMatchMethodPointcutAdvisor;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.ConnectionFactory;
+
+/**
+ * A very simple connection factory that caches a channel per thread. Users are
+ * responsible for releasing the thread's channel by calling
+ * {@link #closeThreadChannel()}.
+ *
+ * @author Gary Russell
+ * @since 2.3
+ *
+ */
+public class ThreadChannelConnectionFactory extends AbstractConnectionFactory {
+
+	private volatile ConnectionWrapper connection;
+
+	private boolean simplePublisherConfirms;
+
+	/**
+	 * Construct an instance.
+	 *
+	 * @param rabbitConnectionFactory the rabbitmq connection factory.
+	 */
+	public ThreadChannelConnectionFactory(ConnectionFactory rabbitConnectionFactory) {
+		this(rabbitConnectionFactory, false);
+	}
+
+	/**
+	 * Construct an instance.
+	 *
+	 * @param rabbitConnectionFactory the rabbitmq connection factory.
+	 * @param isPublisher true if we are creating a publisher connection factory.
+	 */
+	private ThreadChannelConnectionFactory(ConnectionFactory rabbitConnectionFactory, boolean isPublisher) {
+		super(rabbitConnectionFactory);
+		if (!isPublisher) {
+			setPublisherConnectionFactory(new ThreadChannelConnectionFactory(rabbitConnectionFactory, true));
+		}
+	}
+
+	@Override
+	public boolean isSimplePublisherConfirms() {
+		return this.simplePublisherConfirms;
+	}
+
+	/**
+	 * Enable simple publisher confirms.
+	 * @param simplePublisherConfirms true to enable.
+	 */
+	public void setSimplePublisherConfirms(boolean simplePublisherConfirms) {
+		this.simplePublisherConfirms = simplePublisherConfirms;
+	}
+
+	@Override
+	public synchronized Connection createConnection() throws AmqpException {
+		if (this.connection == null || !this.connection.isOpen()) {
+			Connection bareConnection = createBareConnection();
+			this.connection = new ConnectionWrapper(bareConnection.getDelegate(), getCloseTimeout());
+		}
+		return this.connection;
+	}
+
+	/**
+	 * Close the channel associated with this thread, if any.
+	 */
+	public void closeThreadChannel() {
+		ConnectionWrapper connection2 = this.connection;
+		if (connection2 != null) {
+			connection2.closeThreadChannel();
+		}
+	}
+
+	@Override
+	public synchronized void destroy() {
+		super.destroy();
+		if (this.connection != null) {
+			this.connection.forceClose();
+			this.connection = null;
+		}
+	}
+
+	private final class ConnectionWrapper extends SimpleConnection {
+
+		/*
+		 * Intentionally not static.
+		 */
+		private final ThreadLocal<Channel> channels = new ThreadLocal<>();
+
+		private final ThreadLocal<Channel> txChannels = new ThreadLocal<>();
+
+		ConnectionWrapper(com.rabbitmq.client.Connection delegate, int closeTimeout) {
+			super(delegate, closeTimeout);
+		}
+
+		@SuppressWarnings("resource")
+		@Override
+		public Channel createChannel(boolean transactional) {
+			Channel channel = transactional ? this.txChannels.get() : this.channels.get();
+			if (channel == null || !channel.isOpen()) {
+				channel = super.createChannel(transactional);
+				if (transactional) {
+					try {
+						channel.txSelect();
+					}
+					catch (IOException e) {
+						throw RabbitExceptionTranslator.convertRabbitAccessException(e);
+					}
+					channel = createProxy(channel);
+					this.txChannels.set(channel);
+				}
+				else {
+					if (ThreadChannelConnectionFactory.this.simplePublisherConfirms) {
+						try {
+							channel.confirmSelect();
+						}
+						catch (IOException e) {
+							throw RabbitExceptionTranslator.convertRabbitAccessException(e);
+						}
+					}
+					channel = createProxy(channel);
+					this.channels.set(channel);
+				}
+			}
+			return channel;
+		}
+
+		private Channel createProxy(Channel channel) {
+			ProxyFactory pf = new ProxyFactory(channel);
+			Advice advice = new MethodInterceptor() {
+
+				@Override
+				public Object invoke(MethodInvocation invocation) throws Throwable {
+					if (ConnectionWrapper.this.channels.get() == null) {
+						return invocation.proceed();
+					}
+					else {
+						return null;
+					}
+				}
+
+			};
+			NameMatchMethodPointcutAdvisor advisor = new NameMatchMethodPointcutAdvisor(advice);
+			advisor.addMethodName("close");
+			pf.addAdvisor(advisor);
+			return (Channel) pf.getProxy();
+		}
+
+		@Override
+		public void close() {
+		}
+
+		@Override
+		public void closeThreadChannel() {
+			doClose(this.channels);
+			doClose(this.txChannels);
+		}
+
+		private void doClose(ThreadLocal<Channel> channelsTL) {
+			Channel channel = channelsTL.get();
+			if (channel != null) {
+				channelsTL.remove();
+				if (channel.isOpen()) {
+					try {
+						channel.close();
+					}
+					catch (IOException | TimeoutException e) {
+						logger.debug("Error on close", e);
+					}
+				}
+			}
+		}
+
+		void forceClose() {
+			super.close();
+		}
+
+	}
+
+}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java

@@ -62,6 +62,7 @@ import org.springframework.amqp.rabbit.connection.PublisherCallbackChannel;
 import org.springframework.amqp.rabbit.connection.RabbitAccessor;
 import org.springframework.amqp.rabbit.connection.RabbitResourceHolder;
 import org.springframework.amqp.rabbit.connection.RabbitUtils;
+import org.springframework.amqp.rabbit.connection.ThreadChannelConnectionFactory;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder;
@@ -269,7 +270,6 @@ public class RabbitTemplate extends RabbitAccessor // NOSONAR type line count
 	 */
 	public RabbitTemplate() {
 		initDefaultStrategies(); // NOSONAR - intentionally overridable; other assertions will check
-
 	}
 
 	/**
@@ -288,6 +288,14 @@ public class RabbitTemplate extends RabbitAccessor // NOSONAR type line count
 		setMessageConverter(new SimpleMessageConverter());
 	}
 
+	@Override
+	public final void setConnectionFactory(ConnectionFactory connectionFactory) {
+		super.setConnectionFactory(connectionFactory);
+		if (connectionFactory instanceof ThreadChannelConnectionFactory) {
+			this.usePublisherConnection = true;
+		}
+	}
+
 	/**
 	 * The name of the default exchange to use for send operations when none is specified. Defaults to <code>""</code>
 	 * which is the default exchange in the broker (per the AMQP specification).

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitExceptionTranslator.java

@@ -76,7 +76,7 @@ public final class RabbitExceptionTranslator {
 			return new org.springframework.amqp.rabbit.support.ConsumerCancelledException(ex);
 		}
 		if (ex instanceof org.springframework.amqp.rabbit.support.ConsumerCancelledException) {
-			throw (org.springframework.amqp.rabbit.support.ConsumerCancelledException) ex;
+			return (org.springframework.amqp.rabbit.support.ConsumerCancelledException) ex;
 		}
 		// fallback
 		return new UncategorizedAmqpException(ex);

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/PooledChannelConnectionFactoryTests.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbit.connection;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.amqp.rabbit.junit.RabbitAvailable;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.ConnectionFactory;
+
+/**
+ * @author Gary Russell
+ * @since 2.3
+ *
+ */
+@RabbitAvailable
+public class PooledChannelConnectionFactoryTests {
+
+	@Test
+	void testBasic() throws Exception {
+		ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
+		rabbitConnectionFactory.setHost("localhost");
+		PooledChannelConnectionFactory pcf = new PooledChannelConnectionFactory(rabbitConnectionFactory);
+		AtomicBoolean txConfiged = new AtomicBoolean();
+		AtomicBoolean nonTxConfiged = new AtomicBoolean();
+		pcf.setPoolConfigurer((pool, tx) -> {
+			if (tx) {
+				txConfiged.set(true);
+			}
+			else {
+				nonTxConfiged.set(true);
+			}
+		});
+		Connection conn = pcf.createConnection();
+		assertThat(txConfiged.get()).isTrue();
+		assertThat(nonTxConfiged.get()).isTrue();
+		Channel chann1 = conn.createChannel(false);
+		chann1.close();
+		Channel chann2 = conn.createChannel(false);
+		assertThat(chann2).isSameAs(chann1);
+		chann2.close();
+		chann1 = conn.createChannel(true);
+		assertThat(chann1).isNotSameAs(chann2);
+		chann1.close();
+		chann2 = conn.createChannel(true);
+		assertThat(chann2).isSameAs(chann1);
+		pcf.destroy();
+	}
+
+}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ThreadChannelConnectionFactoryTests.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbit.connection;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.amqp.rabbit.junit.RabbitAvailable;
+import org.springframework.amqp.utils.test.TestUtils;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.ConnectionFactory;
+
+/**
+ * @author Gary Russell
+ * @since 2.3
+ *
+ */
+@RabbitAvailable
+public class ThreadChannelConnectionFactoryTests {
+
+	@Test
+	void testBasic() throws Exception {
+		ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
+		rabbitConnectionFactory.setHost("localhost");
+		ThreadChannelConnectionFactory scf = new ThreadChannelConnectionFactory(rabbitConnectionFactory);
+		Connection conn = scf.createConnection();
+		Channel chann1 = conn.createChannel(false);
+		chann1.close();
+		Channel chann2 = conn.createChannel(false);
+		assertThat(chann2).isSameAs(chann1);
+		chann2.close();
+		conn.closeThreadChannel();
+		assertThat(TestUtils.getPropertyValue(conn, "channels", ThreadLocal.class).get()).isNull();
+		chann2 = conn.createChannel(false);
+		assertThat(chann2).isNotSameAs(chann1);
+		chann2.close();
+
+		chann1 = conn.createChannel(true);
+		assertThat(chann1).isNotSameAs(chann2);
+		chann1.close();
+		chann2 = conn.createChannel(true);
+		assertThat(chann2).isSameAs(chann1);
+		chann2.close();
+		assertThat(TestUtils.getPropertyValue(conn, "channels", ThreadLocal.class).get()).isNotNull();
+		assertThat(TestUtils.getPropertyValue(conn, "txChannels", ThreadLocal.class).get()).isNotNull();
+		conn.closeThreadChannel();
+		assertThat(TestUtils.getPropertyValue(conn, "txChannels", ThreadLocal.class).get()).isNull();
+		chann2 = conn.createChannel(true);
+		assertThat(((Channel) TestUtils.getPropertyValue(conn, "txChannels", ThreadLocal.class).get()).isOpen())
+				.isTrue();
+		chann2.close();
+		chann2 = conn.createChannel(false);
+		chann2.close();
+		scf.destroy();
+		assertThat(((Channel) TestUtils.getPropertyValue(conn, "channels", ThreadLocal.class).get()).isOpen())
+				.isFalse();
+		assertThat(((Channel) TestUtils.getPropertyValue(conn, "txChannels", ThreadLocal.class).get()).isOpen())
+				.isFalse();
+	}
+
+}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2019 the original author or authors.
+ * Copyright 2014-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -49,7 +49,7 @@ import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.batch.BatchingStrategy;
 import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy;
-import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ThreadChannelConnectionFactory;
 import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
 import org.springframework.amqp.rabbit.junit.RabbitAvailable;
 import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
@@ -70,6 +70,8 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.util.ReflectionUtils;
 import org.springframework.util.StopWatch;
 
+import com.rabbitmq.client.ConnectionFactory;
+
 /**
  * @author Gary Russell
  * @author Artem Bilan
@@ -84,14 +86,15 @@ public class BatchingRabbitTemplateTests {
 
 	public static final String ROUTE = "test.queue.BatchingRabbitTemplateTests";
 
-	private CachingConnectionFactory connectionFactory;
+	private ThreadChannelConnectionFactory connectionFactory;
 
 	private ThreadPoolTaskScheduler scheduler;
 
 	@BeforeEach
 	public void setup() {
-		this.connectionFactory = new CachingConnectionFactory();
-		this.connectionFactory.setHost("localhost");
+		ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
+		rabbitConnectionFactory.setHost("localhost");
+		this.connectionFactory = new ThreadChannelConnectionFactory(rabbitConnectionFactory);
 		this.connectionFactory.setPort(BrokerTestUtils.getPort());
 		scheduler = new ThreadPoolTaskScheduler();
 		scheduler.setPoolSize(1);

src/reference/asciidoc/amqp.adoc

@@ -201,7 +201,63 @@ Therefore, in this section, we focus on code that exists only within our "`sprin
 
 The central component for managing a connection to the RabbitMQ broker is the `ConnectionFactory` interface.
 The responsibility of a `ConnectionFactory` implementation is to provide an instance of `org.springframework.amqp.rabbit.connection.Connection`, which is a wrapper for `com.rabbitmq.client.Connection`.
-The only concrete implementation we provide is `CachingConnectionFactory`, which, by default, establishes a single connection proxy that can be shared by the application.
+
+[[choosing-factory]]
+===== Choosing a Connection Factory
+
+There are three connection factories to chose from
+
+* `PooledChannelConnectionFactory`
+* `ThreadChannelConnectionFactory`
+* `CachingConnectionFactory`
+
+The first two were added in version 2.3.
+
+For most use cases, the `PooledChannelConnectionFactory` should be used.
+The `ThreadChannelConnectionFactory` can be used if you want to ensure strict message ordering without the need to use <<scoped-operations>>.
+The `CachingConnectionFactory` should be used if you want to use correlated publisher confirmations or if you wish to open multiple connections, via its `CacheMode`.
+
+Simple publisher confirmations are supported by all three factories.
+
+====== `PooledChannelConnectionFactory`
+
+This factory manages a single connection and two pools of channels, based on the Apache Pool2.
+One pool is for transactional channels, the other is for non-transactional channels.
+The pools are `GenericObjectPool` s with default configuration; a callback is provided to configure the pools; refer to the Apache documentation for more information.
+
+The Apache `commons-pool2` jar must be on the class path to use this factory.
+
+====
+[source, java]
+----
+@Bean
+PooledChannelConnectionFactory pcf() throws Exception {
+    ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
+    rabbitConnectionFactory.setHost("localhost");
+    PooledChannelConnectionFactory pcf = new PooledChannelConnectionFactory(rabbitConnectionFactory);
+    pcf.setPoolConfigurer((pool, tx) -> {
+        if (tx) {
+            // configure the transactional pool
+        }
+        else {
+            // configure the non-transactional pool
+        }
+    });
+    return pcf;
+}
+----
+====
+
+====== `ThreadChannelConnectionFactory`
+
+This factory manages a single connection and two `ThreadLocal` s, one for transactional channels, the other for non-transactional channels.
+This factory ensures that all operations on the same thread use the same channel (as long as it remains open).
+This facilitates <<strict-ordering>> without the need for <<scoped-operations>>.
+To avoid memory leaks, if your application uses many short-lived threads, you must call the factory's `closeThreadChannel()` to release the channel resource.
+
+====== `CachingConnectionFactory`
+
+The third implementation provided is the `CachingConnectionFactory`, which, by default, establishes a single connection proxy that can be shared by the application.
 Sharing of the connection is possible since the "`unit of work`" for messaging with AMQP is actually a "`channel`" (in some ways, this is similar to the relationship between a connection and a session in JMS).
 The connection instance provides a `createChannel` method.
 The `CachingConnectionFactory` implementation supports caching of those channels, and it maintains separate caches for channels based on whether they are transactional.
@@ -1038,7 +1094,7 @@ To detect the exception on the sending thread, you can `setChannelTransacted(tru
 However, *transactions significantly impede performance*, so consider this carefully before enabling transactions for just this one use case.
 
 [[template-confirms]]
-===== Publisher Confirms and Returns
+===== Correlated Publisher Confirms and Returns
 
 The `RabbitTemplate` implementation of `AmqpTemplate` supports publisher confirms and returns.
 

src/reference/asciidoc/whats-new.adoc

@@ -6,6 +6,11 @@
 This section describes the changes between version 2.2 and version 2.3.
 See <<change-history>> for changes in previous versions.
 
+==== Connection Factory Changes
+
+Two additional connection factories are now provided.
+See <<choosing-factory>> for more information.
+
 ==== Testing Changes
 
 A new annotation `@SpringRabbitTest` is provided to automatically configure some infrastructure beans for when you are not using `SpringBootTest`.