diff --git a/src/main/java/org/springframework/data/redis/cache/RedisCache.java b/src/main/java/org/springframework/data/redis/cache/RedisCache.java index 476e7da1d..0eae1a3c7 100644 --- a/src/main/java/org/springframework/data/redis/cache/RedisCache.java +++ b/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 getSynchronized(Object key, Callable valueLoader) { + private T getSynchronized(Object key, Callable 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 {@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 loadCacheValue(Object key, Callable 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)); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/ClusterConnectionProvider.java b/src/main/java/org/springframework/data/redis/connection/lettuce/ClusterConnectionProvider.java index 0b4f9efd2..37c9d9669 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/ClusterConnectionProvider.java +++ b/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; - 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); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java index 8e7e2a12b..422b45c18 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java +++ b/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 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(); } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java index f0daa57ce..46c92756f 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java +++ b/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 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(); } } } diff --git a/src/main/java/org/springframework/data/redis/core/script/DefaultRedisScript.java b/src/main/java/org/springframework/data/redis/core/script/DefaultRedisScript.java index 90cb53e71..d8d051b93 100644 --- a/src/main/java/org/springframework/data/redis/core/script/DefaultRedisScript.java +++ b/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 implements RedisScript, 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 implements RedisScript, InitializingBean { public String getSha1() { - synchronized (shaModifiedMonitor) { + lock.lock(); + try { if (sha1 == null || scriptSource.isModified()) { this.sha1 = DigestUtils.sha1DigestAsHex(getScriptAsString()); } return sha1; + } finally { + lock.unlock(); } } diff --git a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java index ea9b8deb9..7d3361391 100644 --- a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java +++ b/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 initialize(BackOffExecution backOffExecution, Collection patterns, Collection channels) { - synchronized (localMonitor) { + lock.lock(); + try { CompletableFuture 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(); } }