Use Locks instead of synchronized blocks that run blocking operations.

To avoid thread pinning on virtual thread arrangements we now use ReentrantLock instead of a synchronized block.

Closes #2690

src/main/java/org/springframework/data/redis/cache/RedisCache.java

@@ -24,6 +24,8 @@ import java.util.Map;
 import java.util.Map.Entry;
 import java.util.StringJoiner;
 import java.util.concurrent.Callable;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.springframework.cache.Cache;
 import org.springframework.cache.support.AbstractValueAdaptingCache;
@@ -58,6 +60,8 @@ public class RedisCache extends AbstractValueAdaptingCache {
 
 	private static final byte[] BINARY_NULL_VALUE = RedisSerializer.java().serialize(NullValue.INSTANCE);
 
+	private final Lock lock = new ReentrantLock();
+
 	private final RedisCacheConfiguration cacheConfiguration;
 
 	private final RedisCacheWriter cacheWriter;
@@ -68,12 +72,12 @@ public class RedisCache extends AbstractValueAdaptingCache {
 	 * Create a new {@link RedisCache}.
 	 *
 	 * @param name {@link String name} for this {@link Cache}; must not be {@literal null}.
-	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations by executing
-	 * the necessary Redis commands; must not be {@literal null}.
-	 * @param cacheConfiguration {@link RedisCacheConfiguration} applied to this {@link RedisCache} on creation;
-	 * must not be {@literal null}.
+	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations by executing the
+	 *          necessary Redis commands; must not be {@literal null}.
+	 * @param cacheConfiguration {@link RedisCacheConfiguration} applied to this {@link RedisCache} on creation; must not
+	 *          be {@literal null}.
 	 * @throws IllegalArgumentException if either the given {@link RedisCacheWriter} or {@link RedisCacheConfiguration}
-	 * are {@literal null} or the given {@link String} name for this {@link RedisCache} is {@literal null}.
+	 *           are {@literal null} or the given {@link String} name for this {@link RedisCache} is {@literal null}.
 	 */
 	protected RedisCache(String name, RedisCacheWriter cacheWriter, RedisCacheConfiguration cacheConfiguration) {
 
@@ -92,7 +96,7 @@ public class RedisCache extends AbstractValueAdaptingCache {
 	 * Get the {@link RedisCacheConfiguration} used to configure this {@link RedisCache} on initialization.
 	 *
 	 * @return an immutable {@link RedisCacheConfiguration} used to configure this {@link RedisCache} on initialization;
-	 * never {@literal null}.
+	 *         never {@literal null}.
 	 */
 	public RedisCacheConfiguration getCacheConfiguration() {
 		return this.cacheConfiguration;
@@ -108,11 +112,11 @@ public class RedisCache extends AbstractValueAdaptingCache {
 	}
 
 	/**
-	 * Gets the configured {@link ConversionService} used to convert {@link Object cache keys} to a {@link String}
-	 * when accessing entries in the cache.
+	 * Gets the configured {@link ConversionService} used to convert {@link Object cache keys} to a {@link String} when
+	 * accessing entries in the cache.
 	 *
-	 * @return the configured {@link ConversionService} used to convert {@link Object cache keys} to a {@link String}
-	 * when accessing entries in the cache.
+	 * @return the configured {@link ConversionService} used to convert {@link Object cache keys} to a {@link String} when
+	 *         accessing entries in the cache.
 	 * @see RedisCacheConfiguration#getConversionService()
 	 * @see #getCacheConfiguration()
 	 */
@@ -153,21 +157,27 @@ public class RedisCache extends AbstractValueAdaptingCache {
 
 	@Nullable
 	@SuppressWarnings("unchecked")
-	private synchronized <T> T getSynchronized(Object key, Callable<T> valueLoader) {
+	private <T> T getSynchronized(Object key, Callable<T> valueLoader) {
 
-		ValueWrapper result = get(key);
+		lock.lock();
+		try {
+			ValueWrapper result = get(key);
+
+			return result != null ? (T) result.get() : loadCacheValue(key, valueLoader);
+		} finally {
+			lock.unlock();
+
+		}
 
-		return result != null ? (T) result.get() : loadCacheValue(key, valueLoader);
 	}
 
 	/**
-	 * Loads the {@link Object} using the given {@link Callable valueLoader} and {@link #put(Object, Object) puts}
-	 * the {@link Object loaded value} in the cache.
+	 * Loads the {@link Object} using the given {@link Callable valueLoader} and {@link #put(Object, Object) puts} the
+	 * {@link Object loaded value} in the cache.
 	 *
 	 * @param <T> {@link Class type} of the loaded {@link Object cache value}.
 	 * @param key {@link Object key} mapped to the loaded {@link Object cache value}.
-	 * @param valueLoader {@link Callable} object used to load the {@link Object value}
-	 * for the given {@link Object key}.
+	 * @param valueLoader {@link Callable} object used to load the {@link Object value} for the given {@link Object key}.
 	 * @return the loaded {@link Object value}.
 	 */
 	protected <T> T loadCacheValue(Object key, Callable<T> valueLoader) {
@@ -176,8 +186,7 @@ public class RedisCache extends AbstractValueAdaptingCache {
 
 		try {
 			value = valueLoader.call();
-		}
-		catch (Exception cause) {
+		} catch (Exception cause) {
 			throw new ValueRetrievalException(key, valueLoader, cause);
 		}
 
@@ -190,8 +199,8 @@ public class RedisCache extends AbstractValueAdaptingCache {
 	protected Object lookup(Object key) {
 
 		byte[] value = getCacheConfiguration().isTimeToIdleEnabled()
-			? getCacheWriter().get(getName(), createAndConvertCacheKey(key), getTimeToLive(key))
-			: getCacheWriter().get(getName(), createAndConvertCacheKey(key));
+				? getCacheWriter().get(getName(), createAndConvertCacheKey(key), getTimeToLive(key))
+				: getCacheWriter().get(getName(), createAndConvertCacheKey(key));
 
 		return value != null ? deserializeCacheValue(value) : null;
 	}
@@ -212,14 +221,14 @@ public class RedisCache extends AbstractValueAdaptingCache {
 		if (nullCacheValueIsNotAllowed(cacheValue)) {
 
 			String message = String.format("Cache '%s' does not allow 'null' values; Avoid storing null"
-				+ " via '@Cacheable(unless=\"#result == null\")' or configure RedisCache to allow 'null'"
-				+ " via RedisCacheConfiguration", getName());
+					+ " via '@Cacheable(unless=\"#result == null\")' or configure RedisCache to allow 'null'"
+					+ " via RedisCacheConfiguration", getName());
 
 			throw new IllegalArgumentException(message);
 		}
 
 		getCacheWriter().put(getName(), createAndConvertCacheKey(key), serializeCacheValue(cacheValue),
-			getTimeToLive(key, value));
+				getTimeToLive(key, value));
 	}
 
 	@Override
@@ -232,7 +241,7 @@ public class RedisCache extends AbstractValueAdaptingCache {
 		}
 
 		byte[] result = getCacheWriter().putIfAbsent(getName(), createAndConvertCacheKey(key),
-			serializeCacheValue(cacheValue), getTimeToLive(key, value));
+				serializeCacheValue(cacheValue), getTimeToLive(key, value));
 
 		return result != null ? new SimpleValueWrapper(fromStoreValue(deserializeCacheValue(result))) : null;
 	}
@@ -313,7 +322,7 @@ public class RedisCache extends AbstractValueAdaptingCache {
 	 *
 	 * @param value array of bytes to deserialize; must not be {@literal null}.
 	 * @return an {@link Object} deserialized from the array of bytes using the configured value
-	 * {@link RedisSerializationContext.SerializationPair}; can be {@literal null}.
+	 *         {@link RedisSerializationContext.SerializationPair}; can be {@literal null}.
 	 * @see RedisCacheConfiguration#getValueSerializationPair()
 	 */
 	@Nullable
@@ -359,8 +368,7 @@ public class RedisCache extends AbstractValueAdaptingCache {
 		if (conversionService.canConvert(source, TypeDescriptor.valueOf(String.class))) {
 			try {
 				return conversionService.convert(key, String.class);
-			}
-			catch (ConversionFailedException cause) {
+			} catch (ConversionFailedException cause) {
 
 				// May fail if the given key is a collection
 				if (isCollectionLikeOrMap(source)) {
@@ -375,8 +383,9 @@ public class RedisCache extends AbstractValueAdaptingCache {
 			return key.toString();
 		}
 
-		String message = String.format("Cannot convert cache key %s to String; Please register a suitable Converter"
-				+ " via 'RedisCacheConfiguration.configureKeyConverters(...)' or override '%s.toString()'",
+		String message = String.format(
+				"Cannot convert cache key %s to String; Please register a suitable Converter"
+						+ " via 'RedisCacheConfiguration.configureKeyConverters(...)' or override '%s.toString()'",
 				source, key.getClass().getName());
 
 		throw new IllegalStateException(message);
@@ -413,13 +422,12 @@ public class RedisCache extends AbstractValueAdaptingCache {
 			target.append("}");
 
 			return target.toString();
-		}
-		else if (source.isCollection() || source.isArray()) {
+		} else if (source.isCollection() || source.isArray()) {
 
 			StringJoiner stringJoiner = new StringJoiner(",");
 
 			Collection<?> collection = source.isCollection() ? (Collection<?>) key
-				: Arrays.asList(ObjectUtils.toObjectArray(key));
+					: Arrays.asList(ObjectUtils.toObjectArray(key));
 
 			for (Object collectedKey : collection) {
 				stringJoiner.add(convertKey(collectedKey));

src/main/java/org/springframework/data/redis/connection/lettuce/ClusterConnectionProvider.java

@@ -25,6 +25,8 @@ import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
 
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.lang.Nullable;
@@ -44,7 +46,7 @@ class ClusterConnectionProvider implements LettuceConnectionProvider, RedisClien
 	private final RedisCodec<?, ?> codec;
 	private final Optional<ReadFrom> readFrom;
 
-	private final Object monitor = new Object();
+	private final Lock lock = new ReentrantLock();
 
 	private volatile boolean initialized;
 
@@ -84,16 +86,19 @@ class ClusterConnectionProvider implements LettuceConnectionProvider, RedisClien
 			// partitions have to be initialized before asynchronous usage.
 			// Needs to happen only once. Initialize eagerly if
 			// blocking is not an options.
-			synchronized (monitor) {
+			lock.lock();
+			try {
 				if (!initialized) {
 					client.getPartitions();
 					initialized = true;
 				}
+			} finally {
+				lock.unlock();
 			}
 		}
 
 		if (connectionType.equals(StatefulRedisPubSubConnection.class)
-			|| connectionType.equals(StatefulRedisClusterPubSubConnection.class)) {
+				|| connectionType.equals(StatefulRedisClusterPubSubConnection.class)) {
 
 			return client.connectPubSubAsync(codec) //
 					.thenApply(connectionType::cast);

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java

@@ -29,6 +29,8 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -551,6 +553,7 @@ public class LettuceClusterConnection extends LettuceConnection
 	 */
 	static class LettuceClusterNodeResourceProvider implements ClusterNodeResourceProvider, DisposableBean {
 
+		private final Lock lock = new ReentrantLock();
 		private final LettuceConnectionProvider connectionProvider;
 		private volatile @Nullable StatefulRedisClusterConnection<byte[], byte[]> connection;
 
@@ -564,11 +567,14 @@ public class LettuceClusterConnection extends LettuceConnection
 
 			Assert.notNull(node, "Node must not be null");
 
-			if (this.connection == null) {
-				synchronized (this) {
-					if (this.connection == null) {
-						this.connection = this.connectionProvider.getConnection(StatefulRedisClusterConnection.class);
+			if (connection == null) {
+				lock.lock();
+				try {
+					if (connection == null) {
+						this.connection = connectionProvider.getConnection(StatefulRedisClusterConnection.class);
 					}
+				} finally {
+					lock.unlock();
 				}
 			}
 

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java

@@ -41,6 +41,8 @@ import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
@@ -139,6 +141,8 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv
 	/** Synchronization monitor for the shared Connection */
 	private final Object connectionMonitor = new Object();
 
+	private final Lock lock = new ReentrantLock();
+
 	private PipeliningFlushPolicy pipeliningFlushPolicy = PipeliningFlushPolicy.flushEachCommand();
 
 	private @Nullable RedisConfiguration configuration;
@@ -1480,7 +1484,8 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv
 		@Nullable
 		StatefulConnection<E, E> getConnection() {
 
-			synchronized (this.connectionMonitor) {
+			lock.lock();
+			try {
 
 				if (this.connection == null) {
 					this.connection = getNativeConnection();
@@ -1491,6 +1496,8 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv
 				}
 
 				return this.connection;
+			} finally {
+				lock.unlock();
 			}
 		}
 
@@ -1508,7 +1515,8 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv
 		 */
 		void validateConnection() {
 
-			synchronized (this.connectionMonitor) {
+			lock.lock();
+			try {
 
 				boolean valid = false;
 
@@ -1535,6 +1543,8 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv
 					resetConnection();
 					this.connection = getNativeConnection();
 				}
+			} finally {
+				lock.unlock();
 			}
 		}
 
@@ -1543,13 +1553,16 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv
 		 */
 		void resetConnection() {
 
-			synchronized (this.connectionMonitor) {
+			lock.lock();
+			try {
 
 				if (this.connection != null) {
 					this.connectionProvider.release(this.connection);
 				}
 
 				this.connection = null;
+			} finally {
+				lock.unlock();
 			}
 		}
 	}

src/main/java/org/springframework/data/redis/core/script/DefaultRedisScript.java

@@ -16,6 +16,8 @@
 package org.springframework.data.redis.core.script;
 
 import java.io.IOException;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.core.io.Resource;
@@ -37,7 +39,7 @@ import org.springframework.util.Assert;
  */
 public class DefaultRedisScript<T> implements RedisScript<T>, InitializingBean {
 
-	private final Object shaModifiedMonitor = new Object();
+	private final Lock lock = new ReentrantLock();
 
 	private @Nullable ScriptSource scriptSource;
 	private @Nullable String sha1;
@@ -76,11 +78,14 @@ public class DefaultRedisScript<T> implements RedisScript<T>, InitializingBean {
 
 	public String getSha1() {
 
-		synchronized (shaModifiedMonitor) {
+		lock.lock();
+		try {
 			if (sha1 == null || scriptSource.isModified()) {
 				this.sha1 = DigestUtils.sha1DigestAsHex(getScriptAsString());
 			}
 			return sha1;
+		} finally {
+			lock.unlock();
 		}
 	}
 

src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java

@@ -32,6 +32,8 @@ import java.util.concurrent.TimeUnit;
 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 java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -1155,7 +1157,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 		private volatile @Nullable RedisConnection connection;
 
 		private final RedisConnectionFactory connectionFactory;
-		private final Object localMonitor = new Object();
+		private final Lock lock = new ReentrantLock();
 		private final DispatchMessageListener delegateListener = new DispatchMessageListener();
 		private final SynchronizingMessageListener synchronizingMessageListener = new SynchronizingMessageListener(
 				delegateListener, delegateListener);
@@ -1176,7 +1178,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 		public CompletableFuture<Void> initialize(BackOffExecution backOffExecution, Collection<byte[]> patterns,
 				Collection<byte[]> channels) {
 
-			synchronized (localMonitor) {
+			lock.lock();
+			try {
 
 				CompletableFuture<Void> initFuture = new CompletableFuture<>();
 				try {
@@ -1200,6 +1203,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 				}
 
 				return initFuture;
+			} finally {
+				lock.unlock();
 			}
 		}
 
@@ -1243,7 +1248,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 
 		public void unsubscribeAll() {
 
-			synchronized (localMonitor) {
+			lock.lock();
+			try {
 
 				RedisConnection connection = this.connection;
 				if (connection == null) {
@@ -1251,6 +1257,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 				}
 
 				doUnsubscribe(connection);
+			} finally {
+				lock.unlock();
 			}
 		}
 
@@ -1266,7 +1274,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 		 */
 		public void cancel() {
 
-			synchronized (localMonitor) {
+			lock.lock();
+			try {
 
 				RedisConnection connection = this.connection;
 				if (connection == null) {
@@ -1274,6 +1283,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 				}
 
 				doCancel(connection);
+			} finally {
+				lock.unlock();
 			}
 		}
 
@@ -1309,7 +1320,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 		 */
 		public void closeConnection() {
 
-			synchronized (localMonitor) {
+			lock.lock();
+			try {
 
 				RedisConnection connection = this.connection;
 				this.connection = null;
@@ -1322,6 +1334,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 						logger.warn("Error closing subscription connection", e);
 					}
 				}
+			} finally {
+				lock.unlock();
 			}
 		}
 
@@ -1367,7 +1381,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 				return;
 			}
 
-			synchronized (localMonitor) {
+			lock.lock();
+			try {
 				RedisConnection connection = this.connection;
 				if (connection != null) {
 					Subscription sub = connection.getSubscription();
@@ -1375,6 +1390,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 						function.accept(sub, data);
 					}
 				}
+			} finally {
+				lock.unlock();
 			}
 		}