From c5489e291249d3da7b8c771b673441099ffcc84f Mon Sep 17 00:00:00 2001 From: Christian Tzolov Date: Mon, 11 Dec 2023 23:22:27 +0100 Subject: [PATCH] GH-2481: Replace trivial synchronized with locks Fixes: #2481 --- .gitignore | 1 + .../amqp/core/AbstractDeclarable.java | 28 ++- .../listener/StreamListenerContainer.java | 125 +++++++---- .../stream/producer/RabbitStreamTemplate.java | 99 ++++++--- .../amqp/rabbit/test/TestRabbitTemplate.java | 21 +- .../amqp/rabbit/AsyncRabbitTemplate.java | 98 +++++---- .../connection/AbstractConnectionFactory.java | 151 +++++++------ .../LocalizedQueueConnectionFactory.java | 31 ++- .../PooledChannelConnectionFactory.java | 43 ++-- .../PublisherCallbackChannelImpl.java | 204 +++++++++++------- .../ThreadChannelConnectionFactory.java | 55 +++-- .../rabbit/core/BatchingRabbitTemplate.java | 59 +++-- .../amqp/rabbit/core/BrokerEventListener.java | 61 ++++-- .../amqp/rabbit/core/RabbitAdmin.java | 5 + .../amqp/rabbit/log4j2/AmqpAppender.java | 12 +- .../RoutingConnectionFactoryTests.java | 44 +++- 16 files changed, 688 insertions(+), 349 deletions(-) diff --git a/.gitignore b/.gitignore index 3b3bb754..3a4003bf 100644 --- a/.gitignore +++ b/.gitignore @@ -20,3 +20,4 @@ nohup.out src/ant/.ant-targets-upload-dist.xml target .sts4-cache +.vscode diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/AbstractDeclarable.java b/spring-amqp/src/main/java/org/springframework/amqp/core/AbstractDeclarable.java index 3f662d14..83002a2e 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/AbstractDeclarable.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/AbstractDeclarable.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2023 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. @@ -22,6 +22,9 @@ import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -30,11 +33,14 @@ import org.springframework.util.Assert; * Base class for {@link Declarable} classes. * * @author Gary Russell + * @author Christian Tzolov * @since 1.2 * */ public abstract class AbstractDeclarable implements Declarable { + private final Lock lock = new ReentrantLock(); + private boolean shouldDeclare = true; private Collection declaringAdmins = new ArrayList(); @@ -109,13 +115,25 @@ public abstract class AbstractDeclarable implements Declarable { } @Override - public synchronized void addArgument(String argName, Object argValue) { - this.arguments.put(argName, argValue); + public void addArgument(String argName, Object argValue) { + this.lock.lock(); + try { + this.arguments.put(argName, argValue); + } + finally { + this.lock.unlock(); + } } @Override - public synchronized Object removeArgument(String name) { - return this.arguments.remove(name); + public Object removeArgument(String name) { + this.lock.lock(); + try { + return this.arguments.remove(name); + } + finally { + this.lock.unlock(); + } } public Map getArguments() { diff --git a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamListenerContainer.java b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamListenerContainer.java index 3f55623d..61775064 100644 --- a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamListenerContainer.java +++ b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamListenerContainer.java @@ -19,6 +19,8 @@ package org.springframework.rabbit.stream.listener; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import org.aopalliance.aop.Advice; import org.apache.commons.logging.LogFactory; @@ -53,6 +55,7 @@ import io.micrometer.observation.ObservationRegistry; * A listener container for RabbitMQ Streams. * * @author Gary Russell + * @author Christian Tzolov * @since 2.4 * */ @@ -60,6 +63,8 @@ public class StreamListenerContainer extends ObservableListenerContainer { protected LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); // NOSONAR + private final Lock lock = new ReentrantLock(); + private final ConsumerBuilder builder; private final Collection consumers = new ArrayList<>(); @@ -111,12 +116,18 @@ public class StreamListenerContainer extends ObservableListenerContainer { * Mutually exclusive with {@link #superStream(String, String)}. */ @Override - public synchronized void setQueueNames(String... queueNames) { + public void setQueueNames(String... queueNames) { Assert.isTrue(!this.superStream, "setQueueNames() and superStream() are mutually exclusive"); Assert.isTrue(queueNames != null && queueNames.length == 1, "Only one stream is supported"); - this.builder.stream(queueNames[0]); - this.simpleStream = true; - this.streamName = queueNames[0]; + this.lock.lock(); + try { + this.builder.stream(queueNames[0]); + this.simpleStream = true; + this.streamName = queueNames[0]; + } + finally { + this.lock.unlock(); + } } /** @@ -139,16 +150,22 @@ public class StreamListenerContainer extends ObservableListenerContainer { * @param consumers the number of consumers. * @since 3.0 */ - public synchronized void superStream(String streamName, String name, int consumers) { - Assert.isTrue(consumers > 0, () -> "'concurrency' must be greater than zero, not " + consumers); - this.concurrency = consumers; - Assert.isTrue(!this.simpleStream, "setQueueNames() and superStream() are mutually exclusive"); - Assert.notNull(streamName, "'superStream' cannot be null"); - this.builder.superStream(streamName) - .singleActiveConsumer() - .name(name); - this.superStream = true; - this.streamName = streamName; + public void superStream(String streamName, String name, int consumers) { + this.lock.lock(); + try { + Assert.isTrue(consumers > 0, () -> "'concurrency' must be greater than zero, not " + consumers); + this.concurrency = consumers; + Assert.isTrue(!this.simpleStream, "setQueueNames() and superStream() are mutually exclusive"); + Assert.notNull(streamName, "'superStream' cannot be null"); + this.builder.superStream(streamName) + .singleActiveConsumer() + .name(name); + this.superStream = true; + this.streamName = streamName; + } + finally { + this.lock.unlock(); + } } /** @@ -176,9 +193,15 @@ public class StreamListenerContainer extends ObservableListenerContainer { * Customize the consumer builder before it is built. * @param consumerCustomizer the customizer. */ - public synchronized void setConsumerCustomizer(ConsumerCustomizer consumerCustomizer) { - Assert.notNull(consumerCustomizer, "'consumerCustomizer' cannot be null"); - this.consumerCustomizer = consumerCustomizer; + public void setConsumerCustomizer(ConsumerCustomizer consumerCustomizer) { + this.lock.lock(); + try { + Assert.notNull(consumerCustomizer, "'consumerCustomizer' cannot be null"); + this.consumerCustomizer = consumerCustomizer; + } + finally { + this.lock.unlock(); + } } @Override @@ -225,36 +248,54 @@ public class StreamListenerContainer extends ObservableListenerContainer { } @Override - public synchronized boolean isRunning() { - return this.consumers.size() > 0; - } - - @Override - public synchronized void start() { - if (this.consumers.size() == 0) { - this.consumerCustomizer.accept(getListenerId(), this.builder); - if (this.simpleStream) { - this.consumers.add(this.builder.build()); - } - else { - for (int i = 0; i < this.concurrency; i++) { - this.consumers.add(this.builder.build()); - } - } + public boolean isRunning() { + this.lock.lock(); + try { + return this.consumers.size() > 0; + } + finally { + this.lock.unlock(); } } @Override - public synchronized void stop() { - this.consumers.forEach(consumer -> { - try { - consumer.close(); + public void start() { + this.lock.lock(); + try { + if (this.consumers.size() == 0) { + this.consumerCustomizer.accept(getListenerId(), this.builder); + if (this.simpleStream) { + this.consumers.add(this.builder.build()); + } + else { + for (int i = 0; i < this.concurrency; i++) { + this.consumers.add(this.builder.build()); + } + } } - catch (RuntimeException ex) { - this.logger.error(ex, "Failed to close consumer"); - } - }); - this.consumers.clear(); + } + finally { + this.lock.unlock(); + } + } + + @Override + public void stop() { + this.lock.lock(); + try { + this.consumers.forEach(consumer -> { + try { + consumer.close(); + } + catch (RuntimeException ex) { + this.logger.error(ex, "Failed to close consumer"); + } + }); + this.consumers.clear(); + } + finally { + this.lock.unlock(); + } } @Override diff --git a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplate.java b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplate.java index 5a8c6357..0751e7db 100644 --- a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplate.java +++ b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplate.java @@ -17,6 +17,8 @@ package org.springframework.rabbit.stream.producer; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import java.util.function.Function; import org.springframework.amqp.core.Message; @@ -52,6 +54,7 @@ import io.micrometer.observation.ObservationRegistry; * Default implementation of {@link RabbitStreamOperations}. * * @author Gary Russell + * @author Christian Tzolov * @since 2.4 * */ @@ -59,6 +62,8 @@ public class RabbitStreamTemplate implements RabbitStreamOperations, Application protected final LogAccessor logger = new LogAccessor(getClass()); // NOSONAR + private final Lock lock = new ReentrantLock(); + private ApplicationContext applicationContext; private final Environment environment; @@ -101,24 +106,30 @@ public class RabbitStreamTemplate implements RabbitStreamOperations, Application } - private synchronized Producer createOrGetProducer() { - if (this.producer == null) { - ProducerBuilder builder = this.environment.producerBuilder(); - if (this.superStreamRouting == null) { - builder.stream(this.streamName); - } - else { - builder.superStream(this.streamName) - .routing(this.superStreamRouting); - } - this.producerCustomizer.accept(this.beanName, builder); - this.producer = builder.build(); - if (!this.streamConverterSet) { - ((DefaultStreamMessageConverter) this.streamConverter).setBuilderSupplier( - () -> this.producer.messageBuilder()); + private Producer createOrGetProducer() { + this.lock.lock(); + try { + if (this.producer == null) { + ProducerBuilder builder = this.environment.producerBuilder(); + if (this.superStreamRouting == null) { + builder.stream(this.streamName); + } + else { + builder.superStream(this.streamName) + .routing(this.superStreamRouting); + } + this.producerCustomizer.accept(this.beanName, builder); + this.producer = builder.build(); + if (!this.streamConverterSet) { + ((DefaultStreamMessageConverter) this.streamConverter).setBuilderSupplier( + () -> this.producer.messageBuilder()); + } } + return this.producer; + } + finally { + this.lock.unlock(); } - return this.producer; } @Override @@ -127,8 +138,14 @@ public class RabbitStreamTemplate implements RabbitStreamOperations, Application } @Override - public synchronized void setBeanName(String name) { - this.beanName = name; + public void setBeanName(String name) { + this.lock.lock(); + try { + this.beanName = name; + } + finally { + this.lock.unlock(); + } } /** @@ -136,8 +153,14 @@ public class RabbitStreamTemplate implements RabbitStreamOperations, Application * @param superStreamRouting the routing function. * @since 3.0 */ - public synchronized void setSuperStreamRouting(Function superStreamRouting) { - this.superStreamRouting = superStreamRouting; + public void setSuperStreamRouting(Function superStreamRouting) { + this.lock.lock(); + try { + this.superStreamRouting = superStreamRouting; + } + finally { + this.lock.unlock(); + } } @@ -155,19 +178,31 @@ public class RabbitStreamTemplate implements RabbitStreamOperations, Application * for {@link #send(Message)} and {@link #convertAndSend(Object)} methods. * @param streamConverter the converter. */ - public synchronized void setStreamConverter(StreamMessageConverter streamConverter) { + public void setStreamConverter(StreamMessageConverter streamConverter) { Assert.notNull(streamConverter, "'streamConverter' cannot be null"); - this.streamConverter = streamConverter; - this.streamConverterSet = true; + this.lock.lock(); + try { + this.streamConverter = streamConverter; + this.streamConverterSet = true; + } + finally { + this.lock.unlock(); + } } /** * Used to customize the {@link ProducerBuilder} before the {@link Producer} is built. * @param producerCustomizer the customizer; */ - public synchronized void setProducerCustomizer(ProducerCustomizer producerCustomizer) { + public void setProducerCustomizer(ProducerCustomizer producerCustomizer) { Assert.notNull(producerCustomizer, "'producerCustomizer' cannot be null"); - this.producerCustomizer = producerCustomizer; + this.lock.lock(); + try { + this.producerCustomizer = producerCustomizer; + } + finally { + this.lock.unlock(); + } } /** @@ -303,10 +338,16 @@ public class RabbitStreamTemplate implements RabbitStreamOperations, Application * operation that requires one. */ @Override - public synchronized void close() { - if (this.producer != null) { - this.producer.close(); - this.producer = null; + public void close() { + this.lock.lock(); + try { + if (this.producer != null) { + this.producer.close(); + this.producer = null; + } + } + finally { + this.lock.unlock(); } } diff --git a/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/TestRabbitTemplate.java b/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/TestRabbitTemplate.java index 246cf0db..c2c74b94 100644 --- a/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/TestRabbitTemplate.java +++ b/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/TestRabbitTemplate.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2023 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. @@ -29,6 +29,8 @@ import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Stream; import org.springframework.amqp.core.Message; @@ -60,6 +62,7 @@ import com.rabbitmq.client.Envelope; * * @author Gary Russell * @author Artem Bilan + * @author Christian Tzolov * * @since 2.0 * @@ -186,6 +189,8 @@ public class TestRabbitTemplate extends RabbitTemplate private static class Listeners { + private final Lock lock = new ReentrantLock(); + private final List listeners = new ArrayList<>(); private volatile Iterator iterator; @@ -193,11 +198,17 @@ public class TestRabbitTemplate extends RabbitTemplate Listeners() { } - private synchronized Object next() { - if (this.iterator == null || !this.iterator.hasNext()) { - this.iterator = this.listeners.iterator(); + private Object next() { + this.lock.lock(); + try { + if (this.iterator == null || !this.iterator.hasNext()) { + this.iterator = this.listeners.iterator(); + } + return this.iterator.next(); + } + finally { + this.lock.unlock(); } - return this.iterator.next(); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java index f6702552..30c3f189 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-2023 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. @@ -22,6 +22,8 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -95,6 +97,8 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa private final Log logger = LogFactory.getLog(this.getClass()); + private final Lock lock = new ReentrantLock(); + private final RabbitTemplate template; private final AbstractMessageListenerContainer container; @@ -337,10 +341,16 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa * @param taskScheduler the task scheduler * @see #setReceiveTimeout(long) */ - public synchronized void setTaskScheduler(TaskScheduler taskScheduler) { + public void setTaskScheduler(TaskScheduler taskScheduler) { Assert.notNull(taskScheduler, "'taskScheduler' cannot be null"); - this.internalTaskScheduler = false; - this.taskScheduler = taskScheduler; + this.lock.lock(); + try { + this.internalTaskScheduler = false; + this.taskScheduler = taskScheduler; + } + finally { + this.lock.unlock(); + } } /** @@ -510,44 +520,56 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa } @Override - public synchronized void start() { - if (!this.running) { - if (this.internalTaskScheduler) { - ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); - scheduler.setThreadNamePrefix(getBeanName() == null ? "asyncTemplate-" : (getBeanName() + "-")); - scheduler.afterPropertiesSet(); - this.taskScheduler = scheduler; - } - if (this.container != null) { - this.container.start(); - } - if (this.directReplyToContainer != null) { - this.directReplyToContainer.setTaskScheduler(this.taskScheduler); - this.directReplyToContainer.start(); + public void start() { + this.lock.lock(); + try { + if (!this.running) { + if (this.internalTaskScheduler) { + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.setThreadNamePrefix(getBeanName() == null ? "asyncTemplate-" : (getBeanName() + "-")); + scheduler.afterPropertiesSet(); + this.taskScheduler = scheduler; + } + if (this.container != null) { + this.container.start(); + } + if (this.directReplyToContainer != null) { + this.directReplyToContainer.setTaskScheduler(this.taskScheduler); + this.directReplyToContainer.start(); + } } + this.running = true; + } + finally { + this.lock.unlock(); } - this.running = true; } @Override - public synchronized void stop() { - if (this.running) { - if (this.container != null) { - this.container.stop(); - } - if (this.directReplyToContainer != null) { - this.directReplyToContainer.stop(); - } - for (RabbitFuture future : this.pending.values()) { - future.setNackCause("AsyncRabbitTemplate was stopped while waiting for reply"); - future.cancel(true); - } - if (this.internalTaskScheduler) { - ((ThreadPoolTaskScheduler) this.taskScheduler).destroy(); - this.taskScheduler = null; + public void stop() { + this.lock.lock(); + try { + if (this.running) { + if (this.container != null) { + this.container.stop(); + } + if (this.directReplyToContainer != null) { + this.directReplyToContainer.stop(); + } + for (RabbitFuture future : this.pending.values()) { + future.setNackCause("AsyncRabbitTemplate was stopped while waiting for reply"); + future.cancel(true); + } + if (this.internalTaskScheduler) { + ((ThreadPoolTaskScheduler) this.taskScheduler).destroy(); + this.taskScheduler = null; + } } + this.running = false; + } + finally { + this.lock.unlock(); } - this.running = false; } @Override @@ -671,7 +693,8 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa @Nullable private ScheduledFuture timeoutTask(RabbitFuture future) { if (this.receiveTimeout > 0) { - synchronized (this) { + this.lock.lock(); + try { if (!this.running) { this.pending.remove(future.getCorrelationId()); throw new IllegalStateException("'AsyncRabbitTemplate' must be started."); @@ -680,6 +703,9 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa new TimeoutTask(future, this.pending, this.directReplyToContainer), Instant.now().plusMillis(this.receiveTimeout)); } + finally { + this.lock.unlock(); + } } return null; } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java index 312e7b20..2acc7e25 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java @@ -32,6 +32,8 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -68,6 +70,7 @@ import com.rabbitmq.client.impl.recovery.AutorecoveringConnection; * @author Steve Powell * @author Artem Bilan * @author Will Droste + * @author Christian Tzolov * */ public abstract class AbstractConnectionFactory implements ConnectionFactory, DisposableBean, BeanNameAware, @@ -80,14 +83,12 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di public enum AddressShuffleMode { /** - * Do not shuffle the addresses before or after opening a connection; attempt - * connections in a fixed order. + * Do not shuffle the addresses before or after opening a connection; attempt connections in a fixed order. */ NONE, /** - * Randomly shuffle the addresses before opening a connection; attempt connections - * in the new order. + * Randomly shuffle the addresses before opening a connection; attempt connections in the new order. */ RANDOM, @@ -106,6 +107,8 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR + private final Lock lock = new ReentrantLock(); + private final com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory; private final CompositeConnectionListener connectionListener = new CompositeConnectionListener(); @@ -144,10 +147,10 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di private int closeTimeout = DEFAULT_CLOSE_TIMEOUT; - private ConnectionNameStrategy connectionNameStrategy = - connectionFactory -> (this.beanName != null ? this.beanName : "SpringAMQP") + - "#" + ObjectUtils.getIdentityHexString(this) + ":" + - this.defaultConnectionNameStrategyCounter.getAndIncrement(); + private ConnectionNameStrategy connectionNameStrategy = connectionFactory -> (this.beanName != null ? this.beanName + : "SpringAMQP") + + "#" + ObjectUtils.getIdentityHexString(this) + ":" + + this.defaultConnectionNameStrategyCounter.getAndIncrement(); private String beanName; @@ -160,8 +163,8 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di private volatile boolean contextStopped; /** - * Create a new AbstractConnectionFactory for the given target ConnectionFactory, - * with no publisher connection factory. + * Create a new AbstractConnectionFactory for the given target ConnectionFactory, with no publisher connection + * factory. * @param rabbitConnectionFactory the target ConnectionFactory */ public AbstractConnectionFactory(com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory) { @@ -170,8 +173,7 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di } /** - * Set a custom publisher connection factory; the type does not need to be the same - * as this factory. + * Set a custom publisher connection factory; the type does not need to be the same as this factory. * @param publisherConnectionFactory the factory. * @since 2.3.2 */ @@ -266,8 +268,8 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di } /** - * Set an {@link AddressResolver} to use when creating connections; overrides - * {@link #setAddresses(String)}, {@link #setHost(String)}, and {@link #setPort(int)}. + * Set an {@link AddressResolver} to use when creating connections; overrides {@link #setAddresses(String)}, + * {@link #setHost(String)}, and {@link #setPort(int)}. * @param addressResolver the resolver. * @since 2.1.15 */ @@ -336,29 +338,40 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di } /** - * Set addresses for clustering. - * This property overrides the host+port properties if not empty. + * Set addresses for clustering. This property overrides the host+port properties if not empty. * @param addresses list of addresses with form "host[:port],..." */ - public synchronized void setAddresses(String addresses) { - if (StringUtils.hasText(addresses)) { - Address[] addressArray = Address.parseAddresses(addresses); - if (addressArray.length > 0) { - this.addresses = new LinkedList<>(Arrays.asList(addressArray)); - if (this.publisherConnectionFactory != null) { - this.publisherConnectionFactory.setAddresses(addresses); + public void setAddresses(String addresses) { + this.lock.lock(); + try { + if (StringUtils.hasText(addresses)) { + Address[] addressArray = Address.parseAddresses(addresses); + if (addressArray.length > 0) { + this.addresses = new LinkedList<>(Arrays.asList(addressArray)); + if (this.publisherConnectionFactory != null) { + this.publisherConnectionFactory.setAddresses(addresses); + } + return; } - return; } + this.logger.info("setAddresses() called with an empty value, will be using the host+port " + + " or addressResolver properties for connections"); + this.addresses = null; + } + finally { + this.lock.unlock(); } - this.logger.info("setAddresses() called with an empty value, will be using the host+port " - + " or addressResolver properties for connections"); - this.addresses = null; } @Nullable - protected synchronized List
getAddresses() throws IOException { - return this.addressResolver != null ? this.addressResolver.getAddresses() : this.addresses; + protected List
getAddresses() throws IOException { + this.lock.lock(); + try { + return this.addressResolver != null ? this.addressResolver.getAddresses() : this.addresses; + } + finally { + this.lock.unlock(); + } } /** @@ -435,10 +448,8 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di } /** - * Provide an Executor for - * use by the Rabbit ConnectionFactory when creating connections. - * Can either be an ExecutorService or a Spring - * ThreadPoolTaskExecutor, as defined by a <task:executor/> element. + * Provide an Executor for use by the Rabbit ConnectionFactory when creating connections. Can either be an + * ExecutorService or a Spring ThreadPoolTaskExecutor, as defined by a <task:executor/> element. * @param executor The executor. */ public void setExecutor(Executor executor) { @@ -463,8 +474,8 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di } /** - * How long to wait (milliseconds) for a response to a connection close - * operation from the broker; default 30000 (30 seconds). + * How long to wait (milliseconds) for a response to a connection close operation from the broker; default 30000 (30 + * seconds). * @param closeTimeout the closeTimeout to set. */ public void setCloseTimeout(int closeTimeout) { @@ -479,8 +490,8 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di } /** - * Provide a {@link ConnectionNameStrategy} to build the name for the target RabbitMQ connection. - * The {@link #beanName} together with a counter is used by default. + * Provide a {@link ConnectionNameStrategy} to build the name for the target RabbitMQ connection. The + * {@link #beanName} together with a counter is used by default. * @param connectionNameStrategy the {@link ConnectionNameStrategy} to use. * @since 2.0 */ @@ -493,10 +504,10 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di } /** - * 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 DEBUG level, since 3.1, previously INFO). + * 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 + * DEBUG level, since 3.1, previously INFO). * @param closeExceptionLogger the {@link ConditionalExceptionLogger}. * @since 1.5 */ @@ -585,7 +596,8 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di } if (this.applicationEventPublisher != null) { - connection.addBlockedListener(new ConnectionBlockedListener(connection, this.applicationEventPublisher)); + connection + .addBlockedListener(new ConnectionBlockedListener(connection, this.applicationEventPublisher)); } return connection; @@ -597,17 +609,23 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di } } - private synchronized com.rabbitmq.client.Connection connect(String connectionName) + private com.rabbitmq.client.Connection connect(String connectionName) throws IOException, TimeoutException { - if (this.addressResolver != null) { - return connectResolver(connectionName); + this.lock.lock(); + try { + if (this.addressResolver != null) { + return connectResolver(connectionName); + } + if (this.addresses != null) { + return connectAddresses(connectionName); + } + else { + return connectHostPort(connectionName); + } } - if (this.addresses != null) { - return connectAddresses(connectionName); - } - else { - return connectHostPort(connectionName); + finally { + this.lock.unlock(); } } @@ -619,22 +637,28 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di connectionName); } - private synchronized com.rabbitmq.client.Connection connectAddresses(String connectionName) + private com.rabbitmq.client.Connection connectAddresses(String connectionName) throws IOException, TimeoutException { - List
addressesToConnect = new ArrayList<>(this.addresses); - if (addressesToConnect.size() > 1 && AddressShuffleMode.RANDOM.equals(this.addressShuffleMode)) { - Collections.shuffle(addressesToConnect); + this.lock.lock(); + try { + List
addressesToConnect = new ArrayList<>(this.addresses); + if (addressesToConnect.size() > 1 && AddressShuffleMode.RANDOM.equals(this.addressShuffleMode)) { + Collections.shuffle(addressesToConnect); + } + if (this.logger.isInfoEnabled()) { + this.logger.info("Attempting to connect to: " + addressesToConnect); + } + com.rabbitmq.client.Connection connection = this.rabbitConnectionFactory.newConnection(this.executorService, + addressesToConnect, connectionName); + if (addressesToConnect.size() > 1 && AddressShuffleMode.INORDER.equals(this.addressShuffleMode)) { + this.addresses.add(this.addresses.remove(0)); + } + return connection; } - if (this.logger.isInfoEnabled()) { - this.logger.info("Attempting to connect to: " + addressesToConnect); + finally { + this.lock.unlock(); } - com.rabbitmq.client.Connection connection = this.rabbitConnectionFactory.newConnection(this.executorService, - addressesToConnect, connectionName); - if (addressesToConnect.size() > 1 && AddressShuffleMode.INORDER.equals(this.addressShuffleMode)) { - this.addresses.add(this.addresses.remove(0)); - } - return connection; } private com.rabbitmq.client.Connection connectHostPort(String connectionName) throws IOException, TimeoutException { @@ -716,8 +740,7 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di } /** - * Default implementation of {@link ConditionalExceptionLogger} for logging channel - * close exceptions. + * Default implementation of {@link ConditionalExceptionLogger} for logging channel close exceptions. * @since 1.5 */ public static class DefaultChannelCloseLogger implements ConditionalExceptionLogger { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactory.java index e83020be..e84ed711 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactory.java @@ -21,6 +21,8 @@ import java.util.Arrays; import java.util.HashMap; import java.util.Map; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -49,6 +51,7 @@ import org.springframework.util.Assert; *

All {@link ConnectionFactory} methods delegate to the default * * @author Gary Russell + * @author Christian Tzolov * @since 1.2 */ public class LocalizedQueueConnectionFactory implements ConnectionFactory, RoutingConnectionFactory, DisposableBean, @@ -56,6 +59,8 @@ public class LocalizedQueueConnectionFactory implements ConnectionFactory, Routi private final Log logger = LogFactory.getLog(getClass()); + private final Lock lock = new ReentrantLock(); + private final Map nodeFactories = new HashMap(); private final ConnectionFactory defaultConnectionFactory; @@ -299,19 +304,25 @@ public class LocalizedQueueConnectionFactory implements ConnectionFactory, Routi return cf; } - private synchronized ConnectionFactory nodeConnectionFactory(String queue, String node, String address) { - if (this.logger.isInfoEnabled()) { - this.logger.info("Queue: " + queue + " is on node: " + node + " at: " + address); - } - ConnectionFactory cf = this.nodeFactories.get(node); - if (cf == null) { - cf = createConnectionFactory(address, node); + private ConnectionFactory nodeConnectionFactory(String queue, String node, String address) { + this.lock.lock(); + try { if (this.logger.isInfoEnabled()) { - this.logger.info("Created new connection factory: " + cf); + this.logger.info("Queue: " + queue + " is on node: " + node + " at: " + address); } - this.nodeFactories.put(node, cf); + ConnectionFactory cf = this.nodeFactories.get(node); + if (cf == null) { + cf = createConnectionFactory(address, node); + if (this.logger.isInfoEnabled()) { + this.logger.info("Created new connection factory: " + cf); + } + this.nodeFactories.put(node, cf); + } + return cf; + } + finally { + this.lock.unlock(); } - return cf; } /** 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 index a1092f1f..b2072f31 100644 --- 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 @@ -20,6 +20,8 @@ import java.io.IOException; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import java.util.function.BiConsumer; import org.aopalliance.aop.Advice; @@ -52,6 +54,7 @@ import com.rabbitmq.client.ShutdownListener; * * @author Gary Russell * @author Leonardo Ferreira + * @author Christian Tzolov * @since 2.3 * */ @@ -60,6 +63,8 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory private final AtomicBoolean running = new AtomicBoolean(); + private final Lock lock = new ReentrantLock(); + private volatile ConnectionWrapper connection; private boolean simplePublisherConfirms; @@ -158,14 +163,20 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory } @Override - public synchronized Connection createConnection() throws AmqpException { - if (this.connection == null || !this.connection.isOpen()) { - Connection bareConnection = createBareConnection(); // NOSONAR - see destroy() - this.connection = new ConnectionWrapper(bareConnection.getDelegate(), getCloseTimeout(), // NOSONAR - this.simplePublisherConfirms, this.poolConfigurer, getChannelListener()); // NOSONAR - getConnectionListener().onCreate(this.connection); + public Connection createConnection() throws AmqpException { + this.lock.lock(); + try { + if (this.connection == null || !this.connection.isOpen()) { + Connection bareConnection = createBareConnection(); // NOSONAR - see destroy() + this.connection = new ConnectionWrapper(bareConnection.getDelegate(), getCloseTimeout(), // NOSONAR + this.simplePublisherConfirms, this.poolConfigurer, getChannelListener()); // NOSONAR + getConnectionListener().onCreate(this.connection); + } + return this.connection; + } + finally { + this.lock.unlock(); } - return this.connection; } /** @@ -180,12 +191,18 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory } @Override - public synchronized void destroy() { - super.destroy(); - if (this.connection != null) { - this.connection.forceClose(); - getConnectionListener().onClose(this.connection); - this.connection = null; + public void destroy() { + this.lock.lock(); + try { + super.destroy(); + if (this.connection != null) { + this.connection.forceClose(); + getConnectionListener().onClose(this.connection); + this.connection = null; + } + } + finally { + this.lock.unlock(); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java index 70ba9312..b0242d86 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2023 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. @@ -34,6 +34,8 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentSkipListMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeoutException; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -88,6 +90,7 @@ import com.rabbitmq.client.impl.recovery.AutorecoveringChannel; * @author Gary Russell * @author Arnaud Cogoluègnes * @author Artem Bilan + * @author Christian Tzolov * * @since 1.0.1 * @@ -99,6 +102,8 @@ public class PublisherCallbackChannelImpl private final Log logger = LogFactory.getLog(this.getClass()); + private final Lock lock = new ReentrantLock(); + private final Channel delegate; private final ConcurrentMap listeners = new ConcurrentHashMap<>(); @@ -127,16 +132,21 @@ public class PublisherCallbackChannelImpl } @Override - public synchronized void setAfterAckCallback(java.util.function.Consumer callback) { - if (getPendingConfirmsCount() == 0 && callback != null) { - callback.accept(this); + public void setAfterAckCallback(java.util.function.Consumer callback) { + this.lock.lock(); + try { + if (getPendingConfirmsCount() == 0 && callback != null) { + callback.accept(this); + } + else { + this.afterAckCallback = callback; + } } - else { - this.afterAckCallback = callback; + finally { + this.lock.unlock(); } } - ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// // BEGIN PURE DELEGATE METHODS ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -721,8 +731,14 @@ public class PublisherCallbackChannelImpl } @Override - public synchronized void clearReturnListeners() { - this.delegate.clearReturnListeners(); + public void clearReturnListeners() { + this.lock.lock(); + try { + this.delegate.clearReturnListeners(); + } + finally { + this.lock.unlock(); + } } @Override @@ -824,42 +840,60 @@ public class PublisherCallbackChannelImpl this.executor.execute(() -> generateNacksForPendingAcks(cause)); } - private synchronized void generateNacksForPendingAcks(String cause) { - for (Entry> entry : this.pendingConfirms.entrySet()) { - Listener listener = entry.getKey(); - for (Entry confirmEntry : entry.getValue().entrySet()) { - confirmEntry.getValue().setCause(cause); - if (this.logger.isDebugEnabled()) { - this.logger.debug(this.toString() + " PC:Nack:(close):" + confirmEntry.getKey()); + private void generateNacksForPendingAcks(String cause) { + this.lock.lock(); + try { + for (Entry> entry : this.pendingConfirms.entrySet()) { + Listener listener = entry.getKey(); + for (Entry confirmEntry : entry.getValue().entrySet()) { + confirmEntry.getValue().setCause(cause); + if (this.logger.isDebugEnabled()) { + this.logger.debug(this.toString() + " PC:Nack:(close):" + confirmEntry.getKey()); + } + processAck(confirmEntry.getKey(), false, false, false); } - processAck(confirmEntry.getKey(), false, false, false); + listener.revoke(this); } - listener.revoke(this); + if (this.logger.isDebugEnabled()) { + this.logger.debug("PendingConfirms cleared"); + } + this.pendingConfirms.clear(); + this.listenerForSeq.clear(); + this.listeners.clear(); } - if (this.logger.isDebugEnabled()) { - this.logger.debug("PendingConfirms cleared"); - } - this.pendingConfirms.clear(); - this.listenerForSeq.clear(); - this.listeners.clear(); - } - - @Override - public synchronized int getPendingConfirmsCount(Listener listener) { - SortedMap pendingConfirmsForListener = this.pendingConfirms.get(listener); - if (pendingConfirmsForListener == null) { - return 0; - } - else { - return pendingConfirmsForListener.entrySet().size(); + finally { + this.lock.unlock(); } } @Override - public synchronized int getPendingConfirmsCount() { - return this.pendingConfirms.values().stream() - .mapToInt(Map::size) - .sum(); + public int getPendingConfirmsCount(Listener listener) { + this.lock.lock(); + try { + SortedMap pendingConfirmsForListener = this.pendingConfirms.get(listener); + if (pendingConfirmsForListener == null) { + return 0; + } + else { + return pendingConfirmsForListener.entrySet().size(); + } + } + finally { + this.lock.unlock(); + } + } + + @Override + public int getPendingConfirmsCount() { + this.lock.lock(); + try { + return this.pendingConfirms.values().stream() + .mapToInt(Map::size) + .sum(); + } + finally { + this.lock.unlock(); + } } /** @@ -882,29 +916,35 @@ public class PublisherCallbackChannelImpl } @Override - public synchronized Collection expire(Listener listener, long cutoffTime) { - SortedMap pendingConfirmsForListener = this.pendingConfirms.get(listener); - if (pendingConfirmsForListener == null) { - return Collections.emptyList(); - } - else { - List expired = new ArrayList(); - Iterator> iterator = pendingConfirmsForListener.entrySet().iterator(); - while (iterator.hasNext()) { - PendingConfirm pendingConfirm = iterator.next().getValue(); - if (pendingConfirm.getTimestamp() < cutoffTime) { - expired.add(pendingConfirm); - iterator.remove(); - CorrelationData correlationData = pendingConfirm.getCorrelationData(); - if (correlationData != null && StringUtils.hasText(correlationData.getId())) { - this.pendingReturns.remove(correlationData.getId()); // NOSONAR never null + public Collection expire(Listener listener, long cutoffTime) { + this.lock.lock(); + try { + SortedMap pendingConfirmsForListener = this.pendingConfirms.get(listener); + if (pendingConfirmsForListener == null) { + return Collections.emptyList(); + } + else { + List expired = new ArrayList(); + Iterator> iterator = pendingConfirmsForListener.entrySet().iterator(); + while (iterator.hasNext()) { + PendingConfirm pendingConfirm = iterator.next().getValue(); + if (pendingConfirm.getTimestamp() < cutoffTime) { + expired.add(pendingConfirm); + iterator.remove(); + CorrelationData correlationData = pendingConfirm.getCorrelationData(); + if (correlationData != null && StringUtils.hasText(correlationData.getId())) { + this.pendingReturns.remove(correlationData.getId()); // NOSONAR never null + } + } + else { + break; } } - else { - break; - } + return expired; } - return expired; + } + finally { + this.lock.unlock(); } } @@ -926,12 +966,18 @@ public class PublisherCallbackChannelImpl processAck(seq, false, multiple, true); } - private synchronized void processAck(long seq, boolean ack, boolean multiple, boolean remove) { + private void processAck(long seq, boolean ack, boolean multiple, boolean remove) { + this.lock.lock(); try { - doProcessAck(seq, ack, multiple, remove); + try { + doProcessAck(seq, ack, multiple, remove); + } + catch (Exception e) { + this.logger.error("Failed to process publisher confirm", e); + } } - catch (Exception e) { - this.logger.error("Failed to process publisher confirm", e); + finally { + this.lock.unlock(); } } @@ -1028,12 +1074,18 @@ public class PublisherCallbackChannelImpl try { if (this.afterAckCallback != null) { java.util.function.Consumer callback = null; - synchronized (this) { + this.lock.lock(); + try { if (getPendingConfirmsCount() == 0) { callback = this.afterAckCallback; this.afterAckCallback = null; } + } + finally { + this.lock.unlock(); + } + if (callback != null) { callback.accept(this); } @@ -1048,18 +1100,24 @@ public class PublisherCallbackChannelImpl } @Override - public synchronized void addPendingConfirm(Listener listener, long seq, PendingConfirm pendingConfirm) { - SortedMap pendingConfirmsForListener = this.pendingConfirms.get(listener); - Assert.notNull(pendingConfirmsForListener, - "Listener not registered: " + listener + " " + this.pendingConfirms.keySet()); - pendingConfirmsForListener.put(seq, pendingConfirm); - this.listenerForSeq.put(seq, listener); - if (pendingConfirm.getCorrelationData() != null) { - String returnCorrelation = pendingConfirm.getCorrelationData().getId(); // NOSONAR never null - if (StringUtils.hasText(returnCorrelation)) { - this.pendingReturns.put(returnCorrelation, pendingConfirm); + public void addPendingConfirm(Listener listener, long seq, PendingConfirm pendingConfirm) { + this.lock.lock(); + try { + SortedMap pendingConfirmsForListener = this.pendingConfirms.get(listener); + Assert.notNull(pendingConfirmsForListener, + "Listener not registered: " + listener + " " + this.pendingConfirms.keySet()); + pendingConfirmsForListener.put(seq, pendingConfirm); + this.listenerForSeq.put(seq, listener); + if (pendingConfirm.getCorrelationData() != null) { + String returnCorrelation = pendingConfirm.getCorrelationData().getId(); // NOSONAR never null + if (StringUtils.hasText(returnCorrelation)) { + this.pendingReturns.put(returnCorrelation, pendingConfirm); + } } } + finally { + this.lock.unlock(); + } } // ReturnListener 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 d31d4413..d1b66753 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 @@ -22,6 +22,8 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; import org.aopalliance.aop.Advice; @@ -46,12 +48,15 @@ import com.rabbitmq.client.ShutdownListener; * * @author Gary Russell * @author Leonardo Ferreira + * @author Christian Tzolov * @since 2.3 * */ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory implements ShutdownListener, SmartLifecycle { + private final Lock lock = new ReentrantLock(); + private final Map contextSwitches = new ConcurrentHashMap<>(); private final Map switchesInProgress = new ConcurrentHashMap<>(); @@ -141,13 +146,19 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory } @Override - public synchronized Connection createConnection() throws AmqpException { - if (this.connection == null || !this.connection.isOpen()) { - Connection bareConnection = createBareConnection(); // NOSONAR - see destroy() - this.connection = new ConnectionWrapper(bareConnection.getDelegate(), getCloseTimeout()); // NOSONAR - getConnectionListener().onCreate(this.connection); + public Connection createConnection() throws AmqpException { + this.lock.lock(); + try { + if (this.connection == null || !this.connection.isOpen()) { + Connection bareConnection = createBareConnection(); // NOSONAR - see destroy() + this.connection = new ConnectionWrapper(bareConnection.getDelegate(), getCloseTimeout()); // NOSONAR + getConnectionListener().onCreate(this.connection); + } + return this.connection; + } + finally { + this.lock.unlock(); } - return this.connection; } /** @@ -172,21 +183,27 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory } @Override - public synchronized void destroy() { - super.destroy(); - if (this.connection != null) { - this.connection.forceClose(); - this.connection = null; + public void destroy() { + this.lock.lock(); + try { + super.destroy(); + if (this.connection != null) { + this.connection.forceClose(); + this.connection = null; + } + if (this.switchesInProgress.size() > 0 && 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(); } - if (this.switchesInProgress.size() > 0 && this.logger.isWarnEnabled()) { - this.logger.warn("Unclaimed context switches from threads:" + - this.switchesInProgress.values() - .stream() - .map(t -> t.getName()) - .collect(Collectors.toList())); + finally { + this.lock.unlock(); } - this.contextSwitches.clear(); - this.switchesInProgress.clear(); } /** diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplate.java index 8316b3e3..8982d809 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplate.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2022 the original author or authors. + * Copyright 2014-2023 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. @@ -18,6 +18,8 @@ package org.springframework.amqp.rabbit.core; import java.util.Date; import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.Message; @@ -43,6 +45,8 @@ import org.springframework.scheduling.TaskScheduler; */ public class BatchingRabbitTemplate extends RabbitTemplate { + private final Lock lock = new ReentrantLock(); + private final BatchingStrategy batchingStrategy; private final TaskScheduler scheduler; @@ -75,27 +79,32 @@ public class BatchingRabbitTemplate extends RabbitTemplate { } @Override - public synchronized void send(String exchange, String routingKey, Message message, + public void send(String exchange, String routingKey, Message message, @Nullable CorrelationData correlationData) throws AmqpException { - - if (correlationData != null) { - if (this.logger.isDebugEnabled()) { - this.logger.debug("Cannot use batching with correlation data"); + this.lock.lock(); + try { + if (correlationData != null) { + if (this.logger.isDebugEnabled()) { + this.logger.debug("Cannot use batching with correlation data"); + } + super.send(exchange, routingKey, message, correlationData); + } + else { + if (this.scheduledTask != null) { + this.scheduledTask.cancel(false); + } + MessageBatch batch = this.batchingStrategy.addToBatch(exchange, routingKey, message); + if (batch != null) { + super.send(batch.getExchange(), batch.getRoutingKey(), batch.getMessage(), null); + } + Date next = this.batchingStrategy.nextRelease(); + if (next != null) { + this.scheduledTask = this.scheduler.schedule((Runnable) () -> releaseBatches(), next.toInstant()); + } } - super.send(exchange, routingKey, message, correlationData); } - else { - if (this.scheduledTask != null) { - this.scheduledTask.cancel(false); - } - MessageBatch batch = this.batchingStrategy.addToBatch(exchange, routingKey, message); - if (batch != null) { - super.send(batch.getExchange(), batch.getRoutingKey(), batch.getMessage(), null); - } - Date next = this.batchingStrategy.nextRelease(); - if (next != null) { - this.scheduledTask = this.scheduler.schedule((Runnable) () -> releaseBatches(), next.toInstant()); - } + finally { + this.lock.unlock(); } } @@ -106,9 +115,15 @@ public class BatchingRabbitTemplate extends RabbitTemplate { releaseBatches(); } - private synchronized void releaseBatches() { - for (MessageBatch batch : this.batchingStrategy.releaseBatches()) { - super.send(batch.getExchange(), batch.getRoutingKey(), batch.getMessage(), null); + private void releaseBatches() { + this.lock.lock(); + try { + for (MessageBatch batch : this.batchingStrategy.releaseBatches()) { + super.send(batch.getExchange(), batch.getRoutingKey(), batch.getMessage(), null); + } + } + finally { + this.lock.unlock(); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BrokerEventListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BrokerEventListener.java index 3b453b7e..8c5113e8 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BrokerEventListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BrokerEventListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2023 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,6 +17,8 @@ package org.springframework.amqp.rabbit.core; import java.util.Arrays; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -50,6 +52,7 @@ import org.springframework.util.ObjectUtils; * with the supplied keys. * * @author Gary Russell + * @author Christian Tzolov * @since 2.1 * */ @@ -58,6 +61,8 @@ public class BrokerEventListener implements MessageListener, ApplicationEventPub private static final Log logger = LogFactory.getLog(BrokerEventListener.class); // NOSONAR - lower case + private final Lock lock = new ReentrantLock(); + private final AbstractMessageListenerContainer container; private final String[] eventKeys; @@ -137,34 +142,52 @@ public class BrokerEventListener implements MessageListener, ApplicationEventPub } @Override - public synchronized void start() { - if (!this.running) { - if (this.stopInvoked) { - // redeclare auto-delete queue - this.stopInvoked = false; - onCreate(null); + public void start() { + this.lock.lock(); + try { + if (!this.running) { + if (this.stopInvoked) { + // redeclare auto-delete queue + this.stopInvoked = false; + onCreate(null); + } + if (this.ownContainer) { + this.container.start(); + } + this.running = true; } - if (this.ownContainer) { - this.container.start(); - } - this.running = true; + } + finally { + this.lock.unlock(); } } @Override - public synchronized void stop() { - if (this.running) { - if (this.ownContainer) { - this.container.stop(); + public void stop() { + this.lock.lock(); + try { + if (this.running) { + if (this.ownContainer) { + this.container.stop(); + } + this.running = false; + this.stopInvoked = true; } - this.running = false; - this.stopInvoked = true; + } + finally { + this.lock.unlock(); } } @Override - public synchronized boolean isRunning() { - return this.running; + public boolean isRunning() { + this.lock.lock(); + try { + return this.running; + } + finally { + this.lock.unlock(); + } } @Override diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java index 5b957a42..2eda92dc 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java @@ -31,6 +31,8 @@ import java.util.Set; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; import org.apache.commons.logging.Log; @@ -79,6 +81,7 @@ import com.rabbitmq.client.Channel; * @author Ed Scriven * @author Gary Russell * @author Artem Bilan + * @author Christian Tzolov */ @ManagedResource(description = "Admin Tasks") public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, ApplicationEventPublisherAware, @@ -122,6 +125,8 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat /** Logger available to subclasses. */ protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR + private final Lock lock = new ReentrantLock(); + private final RabbitTemplate rabbitTemplate; private final Object lifecycleMonitor = new Object(); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java index de488866..d76c38a9 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2020 the original author or authors. + * Copyright 2016-2023 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. @@ -33,6 +33,8 @@ import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import org.apache.logging.log4j.Level; import org.apache.logging.log4j.core.Filter; @@ -139,7 +141,7 @@ public class AmqpAppender extends AbstractAppender { /** * Used to synchronize access to pattern layouts. */ - private final Object layoutMutex = new Object(); + private final Lock layoutMutex = new ReentrantLock(); /** * Construct an instance with the provided properties. @@ -256,10 +258,14 @@ public class AmqpAppender extends AbstractAppender { StringBuilder msgBody; String routingKey; try { - synchronized (this.layoutMutex) { + this.layoutMutex.lock(); + try { msgBody = new StringBuilder(new String(getLayout().toByteArray(logEvent), StandardCharsets.UTF_8)); routingKey = new String(this.manager.routingKeyLayout.toByteArray(logEvent), StandardCharsets.UTF_8); } + finally { + this.layoutMutex.unlock(); + } Message message = null; if (this.manager.charset != null) { try { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RoutingConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RoutingConnectionFactoryTests.java index c22f4b6e..ff5918bc 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RoutingConnectionFactoryTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RoutingConnectionFactoryTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2023 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. @@ -37,6 +37,8 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import org.junit.jupiter.api.Test; import org.mockito.Mockito; @@ -53,6 +55,7 @@ import com.rabbitmq.client.Channel; * @author Josh Chappelle * @author Gary Russell * @author Leonardo Ferreira + * @author Christian Tzolov * @since 1.3 */ public class RoutingConnectionFactoryTests { @@ -224,11 +227,18 @@ public class RoutingConnectionFactoryTests { final AtomicReference connectionMakerKey2 = new AtomicReference<>(); SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory) { - @Override - protected synchronized void redeclareElementsIfNecessary() { - connectionMakerKey2.set(connectionFactory.determineCurrentLookupKey()); - } + Lock lock = new ReentrantLock(); + @Override + protected void redeclareElementsIfNecessary() { + this.lock.lock(); + try { + connectionMakerKey2.set(connectionFactory.determineCurrentLookupKey()); + } + finally { + this.lock.unlock(); + } + } }; container.setQueueNames("foo"); container.setLookupKeyQualifier("xxx"); @@ -263,9 +273,17 @@ public class RoutingConnectionFactoryTests { final AtomicReference connectionMakerKey2 = new AtomicReference<>(); DirectMessageListenerContainer container = new DirectMessageListenerContainer(connectionFactory) { + Lock lock = new ReentrantLock(); + @Override - protected synchronized void redeclareElementsIfNecessary() { - connectionMakerKey2.set(connectionFactory.determineCurrentLookupKey()); + protected void redeclareElementsIfNecessary() { + this.lock.lock(); + try { + connectionMakerKey2.set(connectionFactory.determineCurrentLookupKey()); + } + finally { + this.lock.unlock(); + } } }; @@ -306,9 +324,17 @@ public class RoutingConnectionFactoryTests { final AtomicReference connectionMakerKey2 = new AtomicReference<>(); DirectReplyToMessageListenerContainer container = new DirectReplyToMessageListenerContainer(connectionFactory) { + Lock lock = new ReentrantLock(); + @Override - protected synchronized void redeclareElementsIfNecessary() { - connectionMakerKey2.set(connectionFactory.determineCurrentLookupKey()); + protected void redeclareElementsIfNecessary() { + this.lock.lock(); + try { + connectionMakerKey2.set(connectionFactory.determineCurrentLookupKey()); + } + finally { + this.lock.unlock(); + } } };