Synchronize get with value loader only if value is absent for given key

The previous version synchronize all calls to `get(key, valueLoader)`. After this PR the calls to value loader will only be synchronised if we do not have value for the given key.

Fixes spring-projects#2079

Author: miensol

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

@@ -34,6 +34,7 @@
 import org.springframework.data.redis.util.ByteUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.util.ConcurrentReferenceHashMap;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.ReflectionUtils;
 
@@ -44,6 +45,7 @@
  *
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author Piotr Mionskowski
  * @see RedisCacheConfiguration
  * @see RedisCacheWriter
  * @since 2.0
@@ -118,17 +120,34 @@ public RedisCacheWriter getNativeCache() {
 	 */
 	@Override
 	@SuppressWarnings("unchecked")
-	public synchronized <T> T get(Object key, Callable<T> valueLoader) {
+	public <T> T get(Object key, Callable<T> valueLoader) {
 
 		ValueWrapper result = get(key);
 
 		if (result != null) {
 			return (T) result.get();
 		}
 
-		T value = valueFromLoader(key, valueLoader);
-		put(key, value);
-		return value;
+		return getSynchronized(key, valueLoader);
+	}
+
+	private final ConcurrentReferenceHashMap<String, Object> valueLoaderLocks = new ConcurrentReferenceHashMap<>();
+
+	@SuppressWarnings({"unchecked", "SynchronizationOnLocalVariableOrMethodParameter"})
+	private <T> T getSynchronized(Object key, Callable<T> valueLoader) {
+		final Object loaderLock = valueLoaderLocks.computeIfAbsent(createCacheKey(key), (String k) -> new Object());
+
+		synchronized (loaderLock) {
+			ValueWrapper result = get(key);
+
+			if (result != null) {
+				return (T) result.get();
+			}
+
+			T value = valueFromLoader(key, valueLoader);
+			put(key, value);
+			return value;
+		}
 	}
 
 	/*

src/test/java/org/springframework/data/redis/cache/RedisCacheTests.java

@@ -28,7 +28,12 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Date;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
+import java.util.stream.Stream;
 
 import org.junit.jupiter.api.BeforeEach;
 
@@ -409,15 +414,47 @@ void cacheShouldFailOnNonConvertibleCacheKey() {
 		assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> cache.put(key, sample));
 	}
 
-	void doWithConnection(Consumer<RedisConnection> callback) {
-		RedisConnection connection = connectionFactory.getConnection();
-		try {
-			callback.accept(connection);
-		} finally {
-			connection.close();
-		}
+	@ParameterizedRedisTest // GH-2079
+	void multipleThreadsLoadValueOnce() {
+
+		int threadCount = 5;
+
+		ConcurrentMap<Integer, Integer> valuesByThreadId = new ConcurrentHashMap<>(threadCount);
+
+		CountDownLatch waiter = new CountDownLatch(threadCount);
+
+		AtomicInteger threadIds = new AtomicInteger(0);
+
+		AtomicInteger currentValueForKey = new AtomicInteger(0);
+
+		Stream.generate(threadIds::getAndIncrement)
+				.limit(threadCount)
+				.parallel()
+				.forEach((threadId) -> {
+					waiter.countDown();
+					try {
+						waiter.await();
+					} catch (InterruptedException e) {
+						e.printStackTrace();
+					}
+					Integer valueForThread = cache.get("key", currentValueForKey::incrementAndGet);
+					valuesByThreadId.put(threadId, valueForThread);
+				});
+
+		valuesByThreadId.forEach((thread, valueForThread) -> {
+			assertThat(valueForThread).isEqualTo(currentValueForKey.get());
+		});
 	}
 
+	void doWithConnection(Consumer<RedisConnection> callback) {
+        RedisConnection connection = connectionFactory.getConnection();
+        try {
+            callback.accept(connection);
+        } finally {
+            connection.close();
+        }
+    }
+
 	@Data
 	@NoArgsConstructor
 	@AllArgsConstructor