diff --git a/build.gradle b/build.gradle index 65117fdb..950e801d 100644 --- a/build.gradle +++ b/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") diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/Connection.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/Connection.java index 62985550..04ff8c38 100755 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/Connection.java +++ b/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() { + } + } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PooledChannelConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PooledChannelConnectionFactory.java new file mode 100644 index 00000000..056ecee2 --- /dev/null +++ b/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, 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, 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 channels; + + private final ObjectPool txChannels; + + private final boolean simplePublisherConfirms; + + ConnectionWrapper(com.rabbitmq.client.Connection delegate, int closeTimeout, boolean simplePublisherConfirms, + Log logger, BiConsumer, Boolean> configurer) { + + super(delegate, closeTimeout); + GenericObjectPool 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 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 { + + @Override + public PooledObject 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 p) throws Exception { + p.getObject().close(); + } + + @Override + public boolean validateObject(PooledObject p) { + return p.getObject().isOpen(); + } + + @Override + public void activateObject(PooledObject p) { + } + + @Override + public void passivateObject(PooledObject p) { + } + + } + + private final class TxChannelFactory extends ChannelFactory { + + @Override + public PooledObject 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)); + } + + } + + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitAccessor.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitAccessor.java index fdf52e94..afebee61 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitAccessor.java +++ b/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; } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ThreadChannelConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ThreadChannelConnectionFactory.java new file mode 100644 index 00000000..19bb0094 --- /dev/null +++ b/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 channels = new ThreadLocal<>(); + + private final ThreadLocal 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 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(); + } + + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java index 7b9fccc9..8523edfc 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java +++ b/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 "" * which is the default exchange in the broker (per the AMQP specification). diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitExceptionTranslator.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitExceptionTranslator.java index 20a6a9da..19e339bb 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitExceptionTranslator.java +++ b/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); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/PooledChannelConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/PooledChannelConnectionFactoryTests.java new file mode 100644 index 00000000..b7308a38 --- /dev/null +++ b/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(); + } + +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ThreadChannelConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ThreadChannelConnectionFactoryTests.java new file mode 100644 index 00000000..0a9b6029 --- /dev/null +++ b/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(); + } + +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java index 08b87441..bb4eaaca 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java +++ b/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); diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 7eede797..49a2b6dd 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/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 <>. +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 <> without the need for <>. +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. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 11730a0b..41612a6c 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -6,6 +6,11 @@ This section describes the changes between version 2.2 and version 2.3. See <> for changes in previous versions. +==== Connection Factory Changes + +Two additional connection factories are now provided. +See <> for more information. + ==== Testing Changes A new annotation `@SpringRabbitTest` is provided to automatically configure some infrastructure beans for when you are not using `SpringBootTest`.