Apply non-blocking eviction when using Lettuce for RedisCache.

clear and evict methods now use asynchronous and non-blocking removal of keys when using the Lettuce Redis driver. RedisCache also supports evictIfPresent to remove cache keys immediately which is a blocking method.

Author: mp911de

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

@@ -100,7 +100,7 @@ public long cleanCache(RedisConnection connection, String name, byte[] pattern)
 	 */
 	static class Scan implements BatchStrategy {
 
-		private final int batchSize;
+		final int batchSize;
 
 		Scan(int batchSize) {
 			this.batchSize = batchSize;

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

@@ -27,16 +27,20 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 import org.jspecify.annotations.Nullable;
+
 import org.springframework.dao.PessimisticLockingFailureException;
+import org.springframework.data.redis.connection.ReactiveKeyCommands;
 import org.springframework.data.redis.connection.ReactiveRedisConnection;
 import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
 import org.springframework.data.redis.connection.ReactiveStringCommands;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.RedisStringCommands;
 import org.springframework.data.redis.connection.RedisStringCommands.SetOption;
+import org.springframework.data.redis.core.ScanOptions;
 import org.springframework.data.redis.core.types.Expiration;
 import org.springframework.data.redis.util.ByteUtils;
 import org.springframework.util.Assert;
@@ -311,8 +315,20 @@ public void remove(String name, byte[] key) {
 		Assert.notNull(name, "Name must not be null");
 		Assert.notNull(key, "Key must not be null");
 
-		execute(name, connection -> connection.keyCommands().del(key));
+		if (supportsAsyncRetrieve()) {
+			asyncCacheWriter.remove(name, key).thenRun(() -> statistics.incDeletes(name));
+		} else {
+			removeIfPresent(name, key);
+		}
+	}
+
+	@Override
+	public boolean removeIfPresent(String name, byte[] key) {
+
+		Long removals = execute(name, connection -> connection.keyCommands().del(key));
 		statistics.incDeletes(name);
+
+		return removals > 0;
 	}
 
 	@Override
@@ -321,7 +337,22 @@ public void clean(String name, byte[] pattern) {
 		Assert.notNull(name, "Name must not be null");
 		Assert.notNull(pattern, "Pattern must not be null");
 
-		execute(name, connection -> {
+		if (supportsAsyncRetrieve()) {
+			asyncCacheWriter.clean(name, pattern, batchStrategy)
+					.thenAccept(deleteCount -> statistics.incDeletesBy(name, deleteCount.intValue()));
+			return;
+		}
+
+		invalidate(name, pattern);
+	}
+
+	@Override
+	public boolean invalidate(String name, byte[] pattern) {
+
+		Assert.notNull(name, "Name must not be null");
+		Assert.notNull(pattern, "Pattern must not be null");
+
+		return execute(name, connection -> {
 
 			try {
 				if (isLockingCacheWriter()) {
@@ -337,13 +368,12 @@ public void clean(String name, byte[] pattern) {
 
 				statistics.incDeletesBy(name, (int) deleteCount);
 
+				return deleteCount > 0;
 			} finally {
 				if (isLockingCacheWriter()) {
 					doUnlock(name, connection);
 				}
 			}
-
-			return "OK";
 		});
 	}
 
@@ -499,6 +529,25 @@ interface AsyncCacheWriter {
 		 */
 		CompletableFuture<Void> store(String name, byte[] key, byte[] value, @Nullable Duration ttl);
 
+		/**
+		 * Remove a cache entry asynchronously.
+		 *
+		 * @param name the cache name which to store the cache entry to.
+		 * @param key the key for the cache entry. Must not be {@literal null}.
+		 * @return a future that signals completion.
+		 */
+		CompletableFuture<Void> remove(String name, byte[] key);
+
+		/**
+		 * Clear the cache asynchronously.
+		 *
+		 * @param name the cache name which to store the cache entry to.
+		 * @param pattern {@link String pattern} used to match Redis keys to clear.
+		 * @param batchStrategy strategy to use.
+		 * @return a future that signals completion emitting the number of removed keys.
+		 */
+		CompletableFuture<Long> clean(String name, byte[] pattern, BatchStrategy batchStrategy);
+
 	}
 
 	/**
@@ -524,6 +573,17 @@ public CompletableFuture<byte[]> retrieve(String name, byte[] key, @Nullable Dur
 		public CompletableFuture<Void> store(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
 			throw new UnsupportedOperationException("async store not supported");
 		}
+
+		@Override
+		public CompletableFuture<Void> remove(String name, byte[] key) {
+			throw new UnsupportedOperationException("async remove not supported");
+		}
+
+		@Override
+		public CompletableFuture<Long> clean(String name, byte[] pattern, BatchStrategy batchStrategy) {
+			throw new UnsupportedOperationException("async clean not supported");
+		}
+
 	}
 
 	/**
@@ -534,6 +594,14 @@ public CompletableFuture<Void> store(String name, byte[] key, byte[] value, @Nul
 	 */
 	class AsynchronousCacheWriterDelegate implements AsyncCacheWriter {
 
+		private static final int DEFAULT_SCAN_BATCH_SIZE = 64;
+		private final int cleanBatchSize;
+
+		public AsynchronousCacheWriterDelegate() {
+			this.cleanBatchSize = batchStrategy instanceof BatchStrategies.Scan scan ? scan.batchSize
+					: DEFAULT_SCAN_BATCH_SIZE;
+		}
+
 		@Override
 		public boolean isSupported() {
 			return true;
@@ -561,20 +629,12 @@ public CompletableFuture<Void> store(String name, byte[] key, byte[] value, @Nul
 
 			return doWithConnection(connection -> {
 
-				Mono<?> mono = isLockingCacheWriter() ? doStoreWithLocking(name, key, value, ttl, connection)
-						: doStore(key, value, ttl, connection);
+				Mono<?> mono = doWithLocking(name, key, value, connection, () -> doStore(key, value, ttl, connection));
 
 				return mono.then().toFuture();
 			});
 		}
 
-		private Mono<Boolean> doStoreWithLocking(String name, byte[] key, byte[] value, @Nullable Duration ttl,
-				ReactiveRedisConnection connection) {
-
-			return Mono.usingWhen(doLock(name, key, value, connection), unused -> doStore(key, value, ttl, connection),
-					unused -> doUnlock(name, connection));
-		}
-
 		@SuppressWarnings("NullAway")
 		private Mono<Boolean> doStore(byte[] cacheKey, byte[] value, @Nullable Duration ttl,
 				ReactiveRedisConnection connection) {
@@ -590,6 +650,65 @@ private Mono<Boolean> doStore(byte[] cacheKey, byte[] value, @Nullable Duration
 			}
 		}
 
+		@Override
+		public CompletableFuture<Void> remove(String name, byte[] key) {
+
+			return doWithConnection(connection -> {
+
+				Mono<?> mono = doWithLocking(name, key, null, connection, () -> doRemove(key, connection));
+
+				return mono.then().toFuture();
+			});
+		}
+
+		@Override
+		public CompletableFuture<Long> clean(String name, byte[] pattern, BatchStrategy batchStrategy) {
+
+			return doWithConnection(connection -> {
+
+				Mono<Long> mono = doWithLocking(name, pattern, null, connection, () -> doClean(pattern, connection));
+
+				return mono.toFuture();
+			});
+		}
+
+		private Mono<Long> doClean(byte[] pattern, ReactiveRedisConnection connection) {
+
+			ReactiveKeyCommands commands = connection.keyCommands();
+
+			Flux<ByteBuffer> keys;
+
+			if (batchStrategy instanceof BatchStrategies.Keys) {
+				keys = commands.keys(ByteBuffer.wrap(pattern)).flatMapMany(Flux::fromIterable);
+			} else {
+				keys = commands.scan(ScanOptions.scanOptions().count(cleanBatchSize).match(pattern).build());
+			}
+
+			return keys
+					.buffer(cleanBatchSize) //
+					.flatMap(commands::mUnlink) //
+					.collect(Collectors.summingLong(Long::longValue));
+		}
+
+		@SuppressWarnings("NullAway")
+		private Mono<Long> doRemove(byte[] cacheKey, ReactiveRedisConnection connection) {
+
+			ByteBuffer wrappedKey = ByteBuffer.wrap(cacheKey);
+
+			return connection.keyCommands().unlink(wrappedKey);
+		}
+
+		private <T> Mono<T> doWithLocking(String name, byte[] key, byte @Nullable [] value,
+				ReactiveRedisConnection connection, Supplier<Mono<T>> action) {
+
+			if (isLockingCacheWriter()) {
+				return Mono.usingWhen(doLock(name, key, value, connection), unused -> action.get(),
+						unused -> doUnlock(name, connection));
+			}
+
+			return action.get();
+		}
+
 		private Mono<Object> doLock(String name, Object contextualKey, @Nullable Object contextualValue,
 				ReactiveRedisConnection connection) {
 
@@ -631,5 +750,7 @@ private <T> CompletableFuture<T> doWithConnection(
 					ReactiveRedisConnection::closeLater) //
 					.toFuture();
 		}
+
 	}
+
 }

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

@@ -235,9 +235,12 @@ public void clear() {
 	}
 
 	/**
-	 * Clear keys that match the given {@link String keyPattern}.
+	 * Clear keys that match the given {@link String keyPattern}. Useful when cache keys are formatted in a style where
+	 * Redis patterns can be used for matching these.
 	 * <p>
-	 * Useful when cache keys are formatted in a style where Redis patterns can be used for matching these.
+	 * Actual clearing may be performed in an asynchronous or deferred fashion, with subsequent lookups possibly still
+	 * seeing the entries. This may for example be the case with transactional cache decorators. Use {@link #invalidate()}
+	 * for guaranteed immediate removal of entries.
 	 *
 	 * @param keyPattern {@link String pattern} used to match Redis keys to clear.
 	 * @since 3.0
@@ -246,6 +249,11 @@ public void clear(String keyPattern) {
 		getCacheWriter().clean(getName(), createAndConvertCacheKey(keyPattern));
 	}
 
+	@Override
+	public boolean invalidate() {
+		return getCacheWriter().invalidate(getName(), createAndConvertCacheKey("*"));
+	}
+
 	/**
 	 * Reset all statistics counters and gauges for this cache.
 	 *
@@ -260,6 +268,11 @@ public void evict(Object key) {
 		getCacheWriter().remove(getName(), createAndConvertCacheKey(key));
 	}
 
+	@Override
+	public boolean evictIfPresent(Object key) {
+		return getCacheWriter().removeIfPresent(getName(), createAndConvertCacheKey(key));
+	}
+
 	@Override
 	public CompletableFuture<ValueWrapper> retrieve(Object key) {
 

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

@@ -244,20 +244,54 @@ default CompletableFuture<byte[]> retrieve(String name, byte[] key) {
 
 	/**
 	 * Remove the given key from Redis.
+	 * <p>
+	 * Actual eviction may be performed in an asynchronous or deferred fashion, with subsequent lookups possibly still
+	 * seeing the entry.
 	 *
 	 * @param name cache name must not be {@literal null}.
 	 * @param key key for the cache entry. Must not be {@literal null}.
 	 */
 	void remove(String name, byte[] key);
 
+	/**
+	 * Remove the given key from Redis if it is present, expecting the key to be immediately invisible for subsequent
+	 * lookups.
+	 *
+	 * @param name cache name must not be {@literal null}.
+	 * @param key key for the cache entry. Must not be {@literal null}.
+	 * @return {@code true} if the cache was known to have a mapping for this key before, {@code false} if it did not (or
+	 *         if prior presence could not be determined).
+	 */
+	default boolean removeIfPresent(String name, byte[] key) {
+		remove(name, key);
+		return false;
+	}
+
 	/**
 	 * Remove all keys following the given pattern.
+	 * <p>
+	 * Actual clearing may be performed in an asynchronous or deferred fashion, with subsequent lookups possibly still
+	 * seeing the entries.
 	 *
 	 * @param name cache name must not be {@literal null}.
 	 * @param pattern pattern for the keys to remove. Must not be {@literal null}.
 	 */
 	void clean(String name, byte[] pattern);
 
+	/**
+	 * Remove all keys following the given pattern expecting all entries to be immediately invisible for subsequent
+	 * lookups.
+	 *
+	 * @param name cache name must not be {@literal null}.
+	 * @param pattern pattern for the keys to remove. Must not be {@literal null}.
+	 * @return {@code true} if the cache was known to have mappings before, {@code false} if it did not (or if prior
+	 *         presence of entries could not be determined).
+	 */
+	default boolean invalidate(String name, byte[] pattern) {
+		clean(name, pattern);
+		return false;
+	}
+
 	/**
 	 * Reset all statistics counters and gauges for this cache.
 	 *
@@ -323,4 +357,5 @@ static TtlFunction persistent() {
 		Duration getTimeToLive(Object key, @Nullable Object value);
 
 	}
+
 }