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 index 3d1245b4..5353f0aa 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 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. @@ -17,8 +17,12 @@ package org.springframework.amqp.rabbit.connection; import java.io.IOException; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; import org.aopalliance.aop.Advice; import org.aopalliance.intercept.MethodInterceptor; @@ -28,6 +32,7 @@ import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator; import org.springframework.aop.framework.ProxyFactory; import org.springframework.aop.support.NameMatchMethodPointcutAdvisor; import org.springframework.lang.Nullable; +import org.springframework.util.Assert; import com.rabbitmq.client.Channel; import com.rabbitmq.client.ConnectionFactory; @@ -44,6 +49,10 @@ import com.rabbitmq.client.ShutdownListener; */ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory implements ShutdownListener { + private final Map contextSwitches = new ConcurrentHashMap<>(); + + private final Map switchesInProgress = new ConcurrentHashMap<>(); + private volatile ConnectionWrapper connection; private boolean simplePublisherConfirms; @@ -142,6 +151,79 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory im this.connection.forceClose(); this.connection = null; } + if (this.switchesInProgress.size() > 0) { + if (this.logger.isWarnEnabled()) { + this.logger.warn("Unclaimed context switches from threads:" + + this.switchesInProgress.values() + .stream() + .map(t -> t.getName()) + .collect(Collectors.toList())); + } + } + this.contextSwitches.clear(); + this.switchesInProgress.clear(); + } + + /** + * Call to prepare to switch the channel(s) owned by this thread to another thread. + * @return an opaque object representing the context to switch. If there are no channels + * or no open channels assigned to this thread, null is returned. + * @since 2.3.7 + * @see #switchContext(Object) + */ + @Nullable + public Object prepareSwitchContext() { + return prepareSwitchContext(UUID.randomUUID()); + } + + @Nullable + Object prepareSwitchContext(UUID uuid) { + Object pubContext = null; + if (getPublisherConnectionFactory() instanceof ThreadChannelConnectionFactory) { + pubContext = ((ThreadChannelConnectionFactory) getPublisherConnectionFactory()).prepareSwitchContext(uuid); + } + Context context = ((ConnectionWrapper) createConnection()).prepareSwitchContext(); + if (context.getNonTx() == null && context.getTx() == null) { + this.logger.debug("No channels are bound to this thread"); + return pubContext; + } + if (this.switchesInProgress.values().contains(Thread.currentThread())) { + this.logger + .warn("A previous context switch from this thread has not been claimed yet; possible memory leak?"); + } + this.contextSwitches.put(uuid, context); + this.switchesInProgress.put(uuid, Thread.currentThread()); + return uuid; + } + + /** + * Acquire ownership of another thread's channel(s) after that thread called + * {@link #prepareSwitchContext()}. + * @param toSwitch the context returned by {@link #prepareSwitchContext()}. + * @since 2.3.7 + * @see #prepareSwitchContext() + */ + public void switchContext(@Nullable Object toSwitch) { + if (toSwitch != null) { + Assert.state(doSwitch(toSwitch), () -> "No context to switch for " + toSwitch.toString()); + } + else { + this.logger.debug("Attempted to switch a null context - no channels to acquire"); + } + } + + boolean doSwitch(@Nullable Object toSwitch) { + boolean switched = false; + if (getPublisherConnectionFactory() instanceof ThreadChannelConnectionFactory) { + switched = ((ThreadChannelConnectionFactory) getPublisherConnectionFactory()).doSwitch(toSwitch); + } + Context context = this.contextSwitches.remove(toSwitch); + this.switchesInProgress.remove(toSwitch); + if (context != null) { + ((ConnectionWrapper) createConnection()).switchContext(context); + switched = true; + } + return switched; } private final class ConnectionWrapper extends SimpleConnection { @@ -183,8 +265,8 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory im } this.channels.set(channel); } + getChannelListener().onCreate(channel, transactional); } - getChannelListener().onCreate(channel, transactional); return channel; } @@ -223,7 +305,7 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory im private void handleClose(Channel channel, boolean transactional) { - if (ConnectionWrapper.this.channels.get() == null) { + if (transactional && this.txChannels.get() == null ? true : this.channels.get() == null) { physicalClose(channel); } else { @@ -235,7 +317,6 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory im else { this.channels.remove(); } - RabbitUtils.clearPhysicalCloseRequired(); } } } @@ -266,6 +347,9 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory im catch (IOException | TimeoutException e) { logger.debug("Error on close", e); } + finally { + RabbitUtils.clearPhysicalCloseRequired(); + } } } @@ -274,6 +358,54 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory im getConnectionListener().onClose(this); } + Context prepareSwitchContext() { + Context context = new Context(this.channels.get(), this.txChannels.get()); + this.channels.remove(); + this.txChannels.remove(); + return context; + } + + void switchContext(Context context) { + if (context.getNonTx() != null) { + doSwitch(context.getNonTx(), this.channels); + } + if (context.getTx() != null) { + doSwitch(context.getTx(), this.txChannels); + } + } + + private void doSwitch(Channel channel, ThreadLocal channelTL) { + Channel toClose = channelTL.get(); + if (toClose != null) { + RabbitUtils.setPhysicalCloseRequired(channel, true); + physicalClose(toClose); + } + channelTL.set(channel); + } + + } + + private static class Context { + + private final Channel nonTx; + + private final Channel tx; + + Context(@Nullable Channel nonTx, @Nullable Channel tx) { + this.nonTx = nonTx; + this.tx = tx; + } + + @Nullable + Channel getNonTx() { + return this.nonTx; + } + + @Nullable + Channel getTx() { + return this.tx; + } + } } 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 index 2853a141..bc7a9fb0 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 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. @@ -17,20 +17,39 @@ package org.springframework.amqp.rabbit.connection; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import java.io.IOException; +import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.commons.logging.Log; import org.junit.jupiter.api.Test; +import org.springframework.amqp.core.Message; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.junit.RabbitAvailable; import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition; import org.springframework.amqp.utils.test.TestUtils; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.core.task.TaskExecutor; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import com.rabbitmq.client.AMQP.BasicProperties; import com.rabbitmq.client.Channel; import com.rabbitmq.client.ConnectionFactory; @@ -39,7 +58,7 @@ import com.rabbitmq.client.ConnectionFactory; * @since 2.3 * */ -@RabbitAvailable +@RabbitAvailable(queues = "ThreadChannelConnectionFactoryTests.q1") @SpringJUnitConfig @DirtiesContext public class ThreadChannelConnectionFactoryTests { @@ -48,8 +67,8 @@ public class ThreadChannelConnectionFactoryTests { void testBasic() throws Exception { ConnectionFactory rabbitConnectionFactory = new ConnectionFactory(); rabbitConnectionFactory.setHost("localhost"); - ThreadChannelConnectionFactory scf = new ThreadChannelConnectionFactory(rabbitConnectionFactory); - Connection conn = scf.createConnection(); + ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rabbitConnectionFactory); + Connection conn = tccf.createConnection(); Channel chann1 = conn.createChannel(false); chann1.close(); Channel chann2 = conn.createChannel(false); @@ -78,12 +97,12 @@ public class ThreadChannelConnectionFactoryTests { chann2 = conn.createChannel(false); RabbitUtils.setPhysicalCloseRequired(chann2, true); chann2.close(); - scf.setSimplePublisherConfirms(true); + tccf.setSimplePublisherConfirms(true); chann1 = conn.createChannel(false); assertThat(chann1).isNotSameAs(chann2); assertThat(((ChannelProxy) chann1).isConfirmSelected()).isTrue(); chann1.close(); - scf.destroy(); + tccf.destroy(); assertThat(((Channel) TestUtils.getPropertyValue(conn, "channels", ThreadLocal.class).get()).isOpen()) .isFalse(); assertThat(((Channel) TestUtils.getPropertyValue(conn, "txChannels", ThreadLocal.class).get()).isOpen()) @@ -104,6 +123,227 @@ public class ThreadChannelConnectionFactoryTests { assertThat(config.closed).isTrue(); } + @SuppressWarnings("unchecked") + @Test + void contextSwitch() throws Exception { + ConnectionFactory rabbitConnectionFactory = new ConnectionFactory(); + rabbitConnectionFactory.setHost("localhost"); + rabbitConnectionFactory.setAutomaticRecoveryEnabled(false); + ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rabbitConnectionFactory); + TaskExecutor exec = new SimpleAsyncTaskExecutor(); + AtomicReference nonTx = new AtomicReference<>(); + AtomicReference tx = new AtomicReference<>(); + BlockingQueue context = new LinkedBlockingQueue<>(); + exec.execute(() -> { + Connection conn = tccf.createConnection(); + nonTx.set(conn.createChannel(false)); + tx.set(conn.createChannel(true)); + Object ctx = tccf.prepareSwitchContext(); + assertThat(tccf.prepareSwitchContext()).isNull(); + assertThat(TestUtils.getPropertyValue(conn, "channels", ThreadLocal.class).get()).isNull(); + assertThat(TestUtils.getPropertyValue(conn, "txChannels", ThreadLocal.class).get()).isNull(); + context.add(ctx); + }); + Object ctx = context.poll(10, TimeUnit.SECONDS); + assertThat(ctx).isNotNull(); + tccf.switchContext(ctx); + assertThatIllegalStateException().isThrownBy(() -> tccf.switchContext(ctx)); + Connection conn = tccf.createConnection(); + Channel chann1 = conn.createChannel(false); + assertThat(chann1).isSameAs(nonTx.get()); + Channel chann2 = conn.createChannel(true); + assertThat(chann2).isSameAs(tx.get()); + assertThat(TestUtils.getPropertyValue(tccf, "switchesInProgress", Map.class)).isEmpty(); + tccf.switchContext(null); // test no-op + tccf.destroy(); + } + + @SuppressWarnings("unchecked") + @Test + void contextSwitchMulti() throws Exception { + ConnectionFactory rabbitConnectionFactory = new ConnectionFactory(); + rabbitConnectionFactory.setHost("localhost"); + rabbitConnectionFactory.setAutomaticRecoveryEnabled(false); + ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rabbitConnectionFactory); + TaskExecutor exec = new SimpleAsyncTaskExecutor(); + AtomicReference nonTx = new AtomicReference<>(); + AtomicReference tx = new AtomicReference<>(); + BlockingQueue context = new LinkedBlockingQueue<>(); + CountDownLatch latch = new CountDownLatch(1); + exec.execute(() -> { + Connection conn = tccf.createConnection(); + nonTx.set(conn.createChannel(false)); + tx.set(conn.createChannel(true)); + Object ctx = tccf.prepareSwitchContext(); + assertThat(tccf.prepareSwitchContext()).isNull(); + assertThat(TestUtils.getPropertyValue(conn, "channels", ThreadLocal.class).get()).isNull(); + assertThat(TestUtils.getPropertyValue(conn, "txChannels", ThreadLocal.class).get()).isNull(); + conn.createChannel(false); + context.add(ctx); + context.add(tccf.prepareSwitchContext()); + latch.countDown(); + }); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + Object ctx = context.poll(10, TimeUnit.SECONDS); + assertThat(ctx).isNotNull(); + tccf.switchContext(ctx); + Connection conn = tccf.createConnection(); + Channel chann1 = conn.createChannel(false); + assertThat(chann1).isSameAs(nonTx.get()); + Channel chann2 = conn.createChannel(true); + assertThat(chann2).isSameAs(tx.get()); + assertThat(TestUtils.getPropertyValue(tccf, "switchesInProgress", Map.class)).hasSize(1); + ctx = context.poll(10, TimeUnit.SECONDS); + tccf.switchContext(ctx); + assertThat(TestUtils.getPropertyValue(tccf, "switchesInProgress", Map.class)).isEmpty(); + tccf.destroy(); + } + + @SuppressWarnings("unchecked") + @Test + void contextSwitchBothFactories() throws Exception { + ConnectionFactory rabbitConnectionFactory = new ConnectionFactory(); + rabbitConnectionFactory.setHost("localhost"); + rabbitConnectionFactory.setAutomaticRecoveryEnabled(false); + ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rabbitConnectionFactory); + TaskExecutor exec = new SimpleAsyncTaskExecutor(); + AtomicReference nonTx1 = new AtomicReference<>(); + AtomicReference tx1 = new AtomicReference<>(); + BlockingQueue context = new LinkedBlockingQueue<>(); + AtomicReference nonTx2 = new AtomicReference<>(); + AtomicReference tx2 = new AtomicReference<>(); + exec.execute(() -> { + Connection conn1 = tccf.createConnection(); + nonTx1.set(conn1.createChannel(false)); + tx1.set(conn1.createChannel(true)); + org.springframework.amqp.rabbit.connection.ConnectionFactory pcf = tccf.getPublisherConnectionFactory(); + Connection conn2 = pcf.createConnection(); + nonTx2.set(conn2.createChannel(false)); + tx2.set(conn2.createChannel(true)); + Object ctx = tccf.prepareSwitchContext(); + assertThat(tccf.prepareSwitchContext()).isNull(); + assertThat(TestUtils.getPropertyValue(conn1, "channels", ThreadLocal.class).get()).isNull(); + assertThat(TestUtils.getPropertyValue(conn1, "txChannels", ThreadLocal.class).get()).isNull(); + assertThat(TestUtils.getPropertyValue(conn2, "channels", ThreadLocal.class).get()).isNull(); + assertThat(TestUtils.getPropertyValue(conn2, "txChannels", ThreadLocal.class).get()).isNull(); + context.add(ctx); + }); + Object ctx = context.poll(10, TimeUnit.SECONDS); + assertThat(ctx).isNotNull(); + tccf.switchContext(ctx); + assertThatIllegalStateException().isThrownBy(() -> tccf.switchContext(ctx)); + Connection conn1 = tccf.createConnection(); + Channel chann1 = conn1.createChannel(false); + assertThat(chann1).isSameAs(nonTx1.get()); + Channel chann2 = conn1.createChannel(true); + assertThat(chann2).isSameAs(tx1.get()); + org.springframework.amqp.rabbit.connection.ConnectionFactory pcf = tccf.getPublisherConnectionFactory(); + Connection conn2 = pcf.createConnection(); + Channel chann3 = conn2.createChannel(false); + assertThat(chann3).isSameAs(nonTx2.get()); + Channel chann4 = conn2.createChannel(true); + assertThat(chann4).isSameAs(tx2.get()); + assertThat(TestUtils.getPropertyValue(tccf, "switchesInProgress", Map.class)).isEmpty(); + assertThat(TestUtils.getPropertyValue(tccf, "publisherConnectionFactory.switchesInProgress", Map.class)) + .isEmpty(); + tccf.destroy(); + } + + @Test + void contextSwitchViaTemplate() throws Exception { + // Template uses the nested publisher factory + ConnectionFactory rabbitConnectionFactory = new ConnectionFactory(); + rabbitConnectionFactory.setHost("localhost"); + rabbitConnectionFactory.setAutomaticRecoveryEnabled(false); + ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rabbitConnectionFactory); + RabbitTemplate template = new RabbitTemplate(tccf); + TaskExecutor exec = new SimpleAsyncTaskExecutor(); + AtomicReference nonTx = new AtomicReference<>(); + AtomicReference tx = new AtomicReference<>(); + BlockingQueue context = new LinkedBlockingQueue<>(); + exec.execute(() -> { + template.execute(this::sendChannelNameAsBody); + context.add(tccf.prepareSwitchContext()); + }); + Object ctx = context.poll(10, TimeUnit.SECONDS); + assertThat(ctx).isNotNull(); + tccf.switchContext(ctx); + template.execute(this::sendChannelNameAsBody); + Message received1 = template.receive("ThreadChannelConnectionFactoryTests.q1", 10_000); + Message received2 = template.receive("ThreadChannelConnectionFactoryTests.q1", 10_000); + assertThat(new String(received1.getBody())).isEqualTo(new String(received2.getBody())); + tccf.destroy(); + } + + @Test + void orphanClosed() throws Exception { + ConnectionFactory rabbitConnectionFactory = new ConnectionFactory(); + rabbitConnectionFactory.setHost("localhost"); + rabbitConnectionFactory.setAutomaticRecoveryEnabled(false); + ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rabbitConnectionFactory); + RabbitTemplate template = new RabbitTemplate(tccf); + TaskExecutor exec = new SimpleAsyncTaskExecutor(); + AtomicReference nonTx = new AtomicReference<>(); + AtomicReference tx = new AtomicReference<>(); + BlockingQueue context = new LinkedBlockingQueue<>(); + Runnable task = () -> { + template.execute(channel -> null); + context.add(tccf.prepareSwitchContext()); + }; + exec.execute(task); + Object ctx = context.poll(10, TimeUnit.SECONDS); + assertThat(ctx).isNotNull(); + tccf.switchContext(ctx); + Channel toBeOrphaned = template.execute(channel -> channel); + exec.execute(task); + ctx = context.poll(10, TimeUnit.SECONDS); + assertThat(ctx).isNotNull(); + tccf.switchContext(ctx); + template.execute(channel -> null); + assertThat(toBeOrphaned.isOpen()).isFalse(); + tccf.destroy(); + } + + @Test + void unclaimed() throws Exception { + ConnectionFactory rabbitConnectionFactory = new ConnectionFactory(); + rabbitConnectionFactory.setHost("localhost"); + rabbitConnectionFactory.setAutomaticRecoveryEnabled(false); + ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rabbitConnectionFactory); + Log log = spy(TestUtils.getPropertyValue(tccf, "logger", Log.class)); + new DirectFieldAccessor(tccf).setPropertyValue("logger", log); + given(log.isWarnEnabled()).willReturn(true); + Connection conn = tccf.createConnection(); + conn.createChannel(false); + tccf.prepareSwitchContext(); + tccf.destroy(); + verify(log).warn("Unclaimed context switches from threads:[main]"); + } + + @Test + void neitherBound() throws Exception { + ConnectionFactory rabbitConnectionFactory = new ConnectionFactory(); + rabbitConnectionFactory.setHost("localhost"); + rabbitConnectionFactory.setAutomaticRecoveryEnabled(false); + ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rabbitConnectionFactory); + Log log1 = spy(TestUtils.getPropertyValue(tccf, "logger", Log.class)); + new DirectFieldAccessor(tccf).setPropertyValue("logger", log1); + Log log2 = spy(TestUtils.getPropertyValue(tccf, "publisherConnectionFactory.logger", Log.class)); + new DirectFieldAccessor(tccf).setPropertyValue("publisherConnectionFactory.logger", log2); + given(log1.isDebugEnabled()).willReturn(true); + given(log2.isDebugEnabled()).willReturn(true); + assertThat(tccf.prepareSwitchContext()).isNull(); + verify(log1).debug("No channels are bound to this thread"); + verify(log2).debug("No channels are bound to this thread"); + tccf.destroy(); + } + + Channel sendChannelNameAsBody(Channel channel) throws IOException { + channel.basicPublish("", "ThreadChannelConnectionFactoryTests.q1", new BasicProperties(), + channel.toString().getBytes()); + return channel; + } + @Configuration public static class Config { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java index de881d1f..a8cbcefc 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java @@ -66,7 +66,6 @@ import org.springframework.amqp.rabbit.connection.PublisherCallbackChannel; import org.springframework.amqp.rabbit.connection.RabbitUtils; import org.springframework.amqp.rabbit.connection.SimpleRoutingConnectionFactory; import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; -import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback; import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnsCallback; import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.amqp.utils.SerializationUtils; @@ -576,7 +575,8 @@ public class RabbitTemplateTests { assertThatIllegalStateException().isThrownBy(() -> template.setReturnCallback(mock(RabbitTemplate.ReturnCallback.class))); RabbitTemplate template2 = new RabbitTemplate(); - ReturnCallback callback = mock(RabbitTemplate.ReturnCallback.class); + org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback callback = + mock(org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback.class); template2.setReturnCallback(callback); template2.setReturnCallback(callback); } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 333192aa..a8f30fd0 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -257,6 +257,8 @@ This factory manages a single connection and two `ThreadLocal` s, one for transa This factory ensures that all operations on the same thread use the same channel (as long as it remains open). This facilitates strict message ordering 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. +Starting with version 2.3.7, a thread can transfer its channel(s) to another thread. +See <> for more information. ====== `CachingConnectionFactory` @@ -1386,6 +1388,97 @@ class Service { Even though the publishing is performed on two different threads, they will both use the same channel because the cache is capped at a single channel. +Starting with version 2.3.7, the `ThreadChannelConnectionFactory` supports transferring a thread's channel(s) to another thread, using the `prepareContextSwitch` and `switchContext` methods. +The first method returns a context which is passed to the second thread which calls the second method. +A thread can have either a non-transactional channel or a transactional channel (or one of each) bound to it; you cannot transfer them individually, unless you use two connection factories. +An example follows: + +==== +[source, java] +---- +@SpringBootApplication +public class Application { + + private static final Logger log = LoggerFactory.getLogger(Application.class); + + public static void main(String[] args) { + SpringApplication.run(Application.class, args); + } + + @Bean + TaskExecutor exec() { + ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor(); + exec.setCorePoolSize(10); + return exec; + } + + @Bean + ThreadChannelConnectionFactory tccf() { + ConnectionFactory rabbitConnectionFactory = new ConnectionFactory(); + rabbitConnectionFactory.setHost("localhost"); + return new ThreadChannelConnectionFactory(rabbitConnectionFactory); + } + + @RabbitListener(queues = "queue") + void listen(String in) { + log.info(in); + } + + @Bean + Queue queue() { + return new Queue("queue"); + } + + + @Bean + public ApplicationRunner runner(Service service, TaskExecutor exec) { + return args -> { + exec.execute(() -> service.mainService("test")); + }; + } + +} + +@Component +class Service { + + private static final Logger LOG = LoggerFactory.getLogger(Service.class); + + private final RabbitTemplate template; + + private final TaskExecutor exec; + + private final ThreadChannelConnectionFactory connFactory; + + Service(RabbitTemplate template, TaskExecutor exec, + ThreadChannelConnectionFactory tccf) { + + this.template = template; + this.exec = exec; + this.connFactory = tccf; + } + + void mainService(String toSend) { + LOG.info("Publishing from main service"); + this.template.convertAndSend("queue", toSend); + Object context = this.connFactory.prepareSwitchContext(); + this.exec.execute(() -> secondaryService(toSend.toUpperCase(), context)); + } + + void secondaryService(String toSend, Object threadContext) { + LOG.info("Publishing from secondary service"); + this.connFactory.switchContext(threadContext); + this.template.convertAndSend("queue", toSend); + this.connFactory.closeThreadChannel(); + } + +} +---- +==== + +IMPORTANT: Once the `prepareSwitchContext` is called, if the current thread performs any more operations, they will be performed on a new channel. +It is important to close the thread-bound channel when it is no longer needed. + [[template-messaging]] ===== Messaging Integration