Synchronize RedisCache.get(…) with ValueLoader only if value is absent.

RedisCache.get(…) now optimistically fetches the cache value before entering cache-wide synchronization. The previous version synchronized all calls to `get(key, valueLoader)`.

Closes #2079
Original pull request: #2082.

Author: miensol

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

@@ -44,6 +44,7 @@
  *
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author Piotr Mionskowski
  * @see RedisCacheConfiguration
  * @see RedisCacheWriter
  * @since 2.0
@@ -118,14 +119,25 @@ 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();
 		}
 
+		return getSynchronized(key, valueLoader);
+	}
+
+	@SuppressWarnings("unchecked")
+	private synchronized <T> T getSynchronized(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;

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

@@ -27,7 +27,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;
 
@@ -377,15 +382,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