Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2481: Replace trivial synchronized with locks

Browse Source 
Fixes: #2481
This commit is contained in:
Christian Tzolov
2023-12-11 23:22:27 +01:00
committed by GitHub
parent b7e77fdf1a
commit c5489e2912
16 changed files with 688 additions and 349 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -20,3 +20,4 @@ nohup.out
src/ant/.ant-targets-upload-dist.xml
target
.sts4-cache
.vscode

28
spring-amqp/src/main/java/org/springframework/amqp/core/AbstractDeclarable.java
View File

@@ -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<Object> declaringAdmins = new ArrayList<Object>();
@@ -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<String, Object> getArguments() {

125
spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamListenerContainer.java
View File

@@ -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<Consumer> 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

99
spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplate.java
View File

@@ -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<com.rabbitmq.stream.Message, String> superStreamRouting) {
this.superStreamRouting = superStreamRouting;
public void setSuperStreamRouting(Function<com.rabbitmq.stream.Message, String> 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.</b>
*/
@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();
}
}

21
spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/TestRabbitTemplate.java
View File

@@ -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<Object> listeners = new ArrayList<>();
private volatile Iterator<Object> 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();
}
}

98
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java
View File

@@ -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;
}

151
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java
View File

@@ -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<Address> getAddresses() throws IOException {
return this.addressResolver != null ? this.addressResolver.getAddresses() : this.addresses;
protected List<Address> 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 &lt;task:executor/&gt; 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 &lt;task:executor/&gt; 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<Address> addressesToConnect = new ArrayList<>(this.addresses);
if (addressesToConnect.size() > 1 && AddressShuffleMode.RANDOM.equals(this.addressShuffleMode)) {
Collections.shuffle(addressesToConnect);
this.lock.lock();
try {
List<Address> 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 {

31
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactory.java
View File

@@ -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;
* <p>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<String, ConnectionFactory> nodeFactories = new HashMap<String, ConnectionFactory>();
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;
}
/**

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

@@ -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();
}
}

204
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java
View File

@@ -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<String, Listener> listeners = new ConcurrentHashMap<>();
@@ -127,16 +132,21 @@ public class PublisherCallbackChannelImpl
}
@Override
public synchronized void setAfterAckCallback(java.util.function.Consumer<Channel> callback) {
if (getPendingConfirmsCount() == 0 && callback != null) {
callback.accept(this);
public void setAfterAckCallback(java.util.function.Consumer<Channel> 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<Listener, SortedMap<Long, PendingConfirm>> entry : this.pendingConfirms.entrySet()) {
Listener listener = entry.getKey();
for (Entry<Long, PendingConfirm> 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<Listener, SortedMap<Long, PendingConfirm>> entry : this.pendingConfirms.entrySet()) {
Listener listener = entry.getKey();
for (Entry<Long, PendingConfirm> 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<Long, PendingConfirm> 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<Long, PendingConfirm> 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<PendingConfirm> expire(Listener listener, long cutoffTime) {
SortedMap<Long, PendingConfirm> pendingConfirmsForListener = this.pendingConfirms.get(listener);
if (pendingConfirmsForListener == null) {
return Collections.<PendingConfirm>emptyList();
}
else {
List<PendingConfirm> expired = new ArrayList<PendingConfirm>();
Iterator<Entry<Long, PendingConfirm>> 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<PendingConfirm> expire(Listener listener, long cutoffTime) {
this.lock.lock();
try {
SortedMap<Long, PendingConfirm> pendingConfirmsForListener = this.pendingConfirms.get(listener);
if (pendingConfirmsForListener == null) {
return Collections.<PendingConfirm>emptyList();
}
else {
List<PendingConfirm> expired = new ArrayList<PendingConfirm>();
Iterator<Entry<Long, PendingConfirm>> 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<Channel> 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<Long, PendingConfirm> 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<Long, PendingConfirm> 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

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

@@ -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<UUID, Context> contextSwitches = new ConcurrentHashMap<>();
private final Map<UUID, Thread> 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();
}
/**

59
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplate.java
View File

@@ -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();
}
}

61
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BrokerEventListener.java
View File

@@ -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

5
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java
View File

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

12
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java
View File

@@ -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 {

44
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RoutingConnectionFactoryTests.java
View File

@@ -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<Object> 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<Object> 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<Object> 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();
}
}
};