diff --git a/src/main/java/org/springframework/data/redis/cache/BatchStrategy.java b/src/main/java/org/springframework/data/redis/cache/BatchStrategy.java new file mode 100644 index 000000000..5bf25537e --- /dev/null +++ b/src/main/java/org/springframework/data/redis/cache/BatchStrategy.java @@ -0,0 +1,170 @@ +/* + * Copyright 2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.cache; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.NoSuchElementException; +import java.util.Optional; + +import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.core.Cursor; +import org.springframework.data.redis.core.ScanOptions; +import org.springframework.util.Assert; + +/** + * Batch strategies to be used with {@link RedisCacheWriter}. + *

+ * Primarily used to clear the cache. + * + * @author Mark Paluch + * @since 2.6 + */ +public abstract class BatchStrategy { + + /** + * Batching strategy using a single {@code KEYS} and {@code DEL} command to remove all matching keys. {@code KEYS} + * scans the entire keyspace of the Redis database and can block the Redis worker thread for a long time when the + * keyspace has a significant size. + *

+ * {@code KEYS} is supported for standalone and clustered (sharded) Redis operation modes. + * + * @return batching strategy using {@code KEYS}. + */ + public static BatchStrategy keys() { + return Keys.INSTANCE; + } + + /** + * Batching strategy using a {@code SCAN} cursors and potentially multiple {@code DEL} commands to remove all matching + * keys. This strategy allows a configurable batch size to optimize for scan batching. + *

+ * Note that using the {@code SCAN} strategy might be not supported on all drivers and Redis operation modes. + * + * @return batching strategy using {@code SCAN}. + */ + public static BatchStrategy scan(int batchSize) { + + Assert.isTrue(batchSize > 0, "Batch size must be greater than zero"); + + return new Scan(batchSize); + } + + /** + * Remove all keys following the given pattern. + * + * @param the connection to use. + * @param name The cache name must not be {@literal null}. + * @param pattern The pattern for the keys to remove. Must not be {@literal null}. + * @return number of removed keys. + */ + abstract int cleanCache(RedisConnection connection, String name, byte[] pattern); + + /** + * {@link BatchStrategy} using {@code KEYS}. + */ + static class Keys extends BatchStrategy { + + static Keys INSTANCE = new Keys(); + + @Override + int cleanCache(RedisConnection connection, String name, byte[] pattern) { + + byte[][] keys = Optional.ofNullable(connection.keys(pattern)).orElse(Collections.emptySet()) + .toArray(new byte[0][]); + + if (keys.length > 0) { + connection.del(keys); + } + + return keys.length; + } + } + + /** + * {@link BatchStrategy} using {@code SCAN}. + */ + static class Scan extends BatchStrategy { + + private final int batchSize; + + public Scan(int batchSize) { + this.batchSize = batchSize; + } + + @Override + int cleanCache(RedisConnection connection, String name, byte[] pattern) { + + Cursor cursor = connection.scan(ScanOptions.scanOptions().count(batchSize).match(pattern).build()); + + PartitionIterator partitions = new PartitionIterator<>(cursor, batchSize); + + int count = 0; + + while (partitions.hasNext()) { + + List keys = partitions.next(); + count += keys.size(); + + if (keys.size() > 0) { + connection.del(keys.toArray(new byte[0][])); + } + } + + return count; + } + } + + /** + * Utility to split and buffer outcome from a {@link Iterator} into {@link List lists} of {@code T} with a maximum + * chunks {@code size}. + * + * @param + */ + static class PartitionIterator implements Iterator> { + + private final Iterator iterator; + private final int size; + + public PartitionIterator(Iterator iterator, int size) { + this.iterator = iterator; + this.size = size; + } + + @Override + public boolean hasNext() { + return iterator.hasNext(); + } + + @Override + public List next() { + + if (!hasNext()) { + throw new NoSuchElementException(); + } + + List list = new ArrayList<>(size); + while (list.size() < size && iterator.hasNext()) { + list.add(iterator.next()); + } + + return list; + } + } + +} diff --git a/src/main/java/org/springframework/data/redis/cache/DefaultRedisCacheWriter.java b/src/main/java/org/springframework/data/redis/cache/DefaultRedisCacheWriter.java index 61530140c..6cfe7b950 100644 --- a/src/main/java/org/springframework/data/redis/cache/DefaultRedisCacheWriter.java +++ b/src/main/java/org/springframework/data/redis/cache/DefaultRedisCacheWriter.java @@ -17,8 +17,6 @@ package org.springframework.data.redis.cache; import java.nio.charset.StandardCharsets; import java.time.Duration; -import java.util.Collections; -import java.util.Optional; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import java.util.function.Function; @@ -53,21 +51,24 @@ class DefaultRedisCacheWriter implements RedisCacheWriter { private final RedisConnectionFactory connectionFactory; private final Duration sleepTime; private final CacheStatisticsCollector statistics; + private final BatchStrategy batchStrategy; /** * @param connectionFactory must not be {@literal null}. + * @param batchStrategy must not be {@literal null}. */ - DefaultRedisCacheWriter(RedisConnectionFactory connectionFactory) { - this(connectionFactory, Duration.ZERO); + DefaultRedisCacheWriter(RedisConnectionFactory connectionFactory, BatchStrategy batchStrategy) { + this(connectionFactory, Duration.ZERO, batchStrategy); } /** * @param connectionFactory must not be {@literal null}. * @param sleepTime sleep time between lock request attempts. Must not be {@literal null}. Use {@link Duration#ZERO} * to disable locking. + * @param batchStrategy must not be {@literal null}. */ - DefaultRedisCacheWriter(RedisConnectionFactory connectionFactory, Duration sleepTime) { - this(connectionFactory, sleepTime, CacheStatisticsCollector.none()); + DefaultRedisCacheWriter(RedisConnectionFactory connectionFactory, Duration sleepTime, BatchStrategy batchStrategy) { + this(connectionFactory, sleepTime, CacheStatisticsCollector.none(), batchStrategy); } /** @@ -75,17 +76,20 @@ class DefaultRedisCacheWriter implements RedisCacheWriter { * @param sleepTime sleep time between lock request attempts. Must not be {@literal null}. Use {@link Duration#ZERO} * to disable locking. * @param cacheStatisticsCollector must not be {@literal null}. + * @param batchStrategy must not be {@literal null}. */ DefaultRedisCacheWriter(RedisConnectionFactory connectionFactory, Duration sleepTime, - CacheStatisticsCollector cacheStatisticsCollector) { + CacheStatisticsCollector cacheStatisticsCollector, BatchStrategy batchStrategy) { Assert.notNull(connectionFactory, "ConnectionFactory must not be null!"); Assert.notNull(sleepTime, "SleepTime must not be null!"); Assert.notNull(cacheStatisticsCollector, "CacheStatisticsCollector must not be null!"); + Assert.notNull(batchStrategy, "BatchStrategy must not be null!"); this.connectionFactory = connectionFactory; this.sleepTime = sleepTime; this.statistics = cacheStatisticsCollector; + this.batchStrategy = batchStrategy; } /* @@ -213,13 +217,9 @@ class DefaultRedisCacheWriter implements RedisCacheWriter { wasLocked = true; } - byte[][] keys = Optional.ofNullable(connection.keys(pattern)).orElse(Collections.emptySet()) - .toArray(new byte[0][]); - if (keys.length > 0) { - statistics.incDeletesBy(name, keys.length); - connection.del(keys); - } + statistics.incDeletesBy(name, batchStrategy.cleanCache(connection, name, pattern)); + } finally { if (wasLocked && isLockingCacheWriter()) { @@ -255,7 +255,7 @@ class DefaultRedisCacheWriter implements RedisCacheWriter { */ @Override public RedisCacheWriter withStatisticsCollector(CacheStatisticsCollector cacheStatisticsCollector) { - return new DefaultRedisCacheWriter(connectionFactory, sleepTime, cacheStatisticsCollector); + return new DefaultRedisCacheWriter(connectionFactory, sleepTime, cacheStatisticsCollector, this.batchStrategy); } /** diff --git a/src/main/java/org/springframework/data/redis/cache/RedisCacheManager.java b/src/main/java/org/springframework/data/redis/cache/RedisCacheManager.java index 5b6718e35..bd764f5d2 100644 --- a/src/main/java/org/springframework/data/redis/cache/RedisCacheManager.java +++ b/src/main/java/org/springframework/data/redis/cache/RedisCacheManager.java @@ -186,7 +186,7 @@ public class RedisCacheManager extends AbstractTransactionSupportingCacheManager Assert.notNull(connectionFactory, "ConnectionFactory must not be null!"); - return new RedisCacheManager(new DefaultRedisCacheWriter(connectionFactory), + return new RedisCacheManager(RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory), RedisCacheConfiguration.defaultCacheConfig()); } @@ -311,7 +311,7 @@ public class RedisCacheManager extends AbstractTransactionSupportingCacheManager Assert.notNull(connectionFactory, "ConnectionFactory must not be null!"); - return new RedisCacheManagerBuilder(new DefaultRedisCacheWriter(connectionFactory)); + return new RedisCacheManagerBuilder(RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory)); } /** diff --git a/src/main/java/org/springframework/data/redis/cache/RedisCacheWriter.java b/src/main/java/org/springframework/data/redis/cache/RedisCacheWriter.java index 8d5d135cf..abe121ee8 100644 --- a/src/main/java/org/springframework/data/redis/cache/RedisCacheWriter.java +++ b/src/main/java/org/springframework/data/redis/cache/RedisCacheWriter.java @@ -26,6 +26,9 @@ import org.springframework.util.Assert; * caching.
* The {@link RedisCacheWriter} may be shared by multiple cache implementations and is responsible for writing / reading * binary data to / from Redis. The implementation honors potential cache lock flags that might be set. + *

+ * The default {@link RedisCacheWriter} implementation can be customized with {@link BatchStrategy} to tune performance + * behavior. * * @author Christoph Strobl * @author Mark Paluch @@ -40,10 +43,24 @@ public interface RedisCacheWriter extends CacheStatisticsProvider { * @return new instance of {@link DefaultRedisCacheWriter}. */ static RedisCacheWriter nonLockingRedisCacheWriter(RedisConnectionFactory connectionFactory) { + return nonLockingRedisCacheWriter(connectionFactory, BatchStrategy.keys()); + } + + /** + * Create new {@link RedisCacheWriter} without locking behavior. + * + * @param connectionFactory must not be {@literal null}. + * @param batchStrategy must not be {@literal null}. + * @return new instance of {@link DefaultRedisCacheWriter}. + * @since 2.6 + */ + static RedisCacheWriter nonLockingRedisCacheWriter(RedisConnectionFactory connectionFactory, + BatchStrategy batchStrategy) { Assert.notNull(connectionFactory, "ConnectionFactory must not be null!"); + Assert.notNull(batchStrategy, "BatchStrategy must not be null!"); - return new DefaultRedisCacheWriter(connectionFactory); + return new DefaultRedisCacheWriter(connectionFactory, batchStrategy); } /** @@ -53,10 +70,23 @@ public interface RedisCacheWriter extends CacheStatisticsProvider { * @return new instance of {@link DefaultRedisCacheWriter}. */ static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectionFactory) { + return lockingRedisCacheWriter(connectionFactory, BatchStrategy.keys()); + } + + /** + * Create new {@link RedisCacheWriter} with locking behavior. + * + * @param connectionFactory must not be {@literal null}. + * @param batchStrategy must not be {@literal null}. + * @return new instance of {@link DefaultRedisCacheWriter}. + * @since 2.6 + */ + static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectionFactory, + BatchStrategy batchStrategy) { Assert.notNull(connectionFactory, "ConnectionFactory must not be null!"); - return new DefaultRedisCacheWriter(connectionFactory, Duration.ofMillis(50)); + return new DefaultRedisCacheWriter(connectionFactory, Duration.ofMillis(50), batchStrategy); } /** diff --git a/src/test/java/org/springframework/data/redis/cache/DefaultRedisCacheWriterTests.java b/src/test/java/org/springframework/data/redis/cache/DefaultRedisCacheWriterTests.java index 8310bdb45..fa3536f28 100644 --- a/src/test/java/org/springframework/data/redis/cache/DefaultRedisCacheWriterTests.java +++ b/src/test/java/org/springframework/data/redis/cache/DefaultRedisCacheWriterTests.java @@ -306,7 +306,8 @@ public class DefaultRedisCacheWriterTests { Thread th = new Thread(() -> { - DefaultRedisCacheWriter writer = new DefaultRedisCacheWriter(connectionFactory, Duration.ofMillis(50)) { + DefaultRedisCacheWriter writer = new DefaultRedisCacheWriter(connectionFactory, Duration.ofMillis(50), + BatchStrategy.keys()) { @Override boolean doCheckLock(String name, RedisConnection connection) { diff --git a/src/test/java/org/springframework/data/redis/cache/LegacyRedisCacheTests.java b/src/test/java/org/springframework/data/redis/cache/LegacyRedisCacheTests.java index 5dd970712..46f93bb5d 100644 --- a/src/test/java/org/springframework/data/redis/cache/LegacyRedisCacheTests.java +++ b/src/test/java/org/springframework/data/redis/cache/LegacyRedisCacheTests.java @@ -102,7 +102,8 @@ public class LegacyRedisCacheTests { cacheConfiguration = cacheConfiguration.disableCachingNullValues(); } - return new RedisCache(CACHE_NAME, new DefaultRedisCacheWriter(connectionFactory), cacheConfiguration); + return new RedisCache(CACHE_NAME, RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory), + cacheConfiguration); } protected Object getValue() { diff --git a/src/test/java/org/springframework/data/redis/cache/RedisCacheTests.java b/src/test/java/org/springframework/data/redis/cache/RedisCacheTests.java index 3e5fce8a7..a32d7c4bf 100644 --- a/src/test/java/org/springframework/data/redis/cache/RedisCacheTests.java +++ b/src/test/java/org/springframework/data/redis/cache/RedisCacheTests.java @@ -16,6 +16,7 @@ package org.springframework.data.redis.cache; import static org.assertj.core.api.Assertions.*; +import static org.assertj.core.api.Assumptions.*; import lombok.AllArgsConstructor; import lombok.Data; @@ -37,6 +38,7 @@ import org.springframework.cache.interceptor.SimpleKeyGenerator; import org.springframework.cache.support.NullValue; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair; import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.data.redis.test.extension.parametrized.MethodSource; @@ -81,7 +83,7 @@ public class RedisCacheTests { doWithConnection(RedisConnection::flushAll); - cache = new RedisCache("cache", new DefaultRedisCacheWriter(connectionFactory), + cache = new RedisCache("cache", RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory), RedisCacheConfiguration.defaultCacheConfig().serializeValuesWith(SerializationPair.fromSerializer(serializer))); } @@ -251,6 +253,33 @@ public class RedisCacheTests { }); } + @ParameterizedRedisTest // GH-1721 + void clearWithScanShouldClearCache() { + + // SCAN not supported via Jedis Cluster. + if (connectionFactory instanceof JedisConnectionFactory) { + assumeThat(((JedisConnectionFactory) connectionFactory).isRedisClusterAware()).isFalse(); + } + + RedisCache cache = new RedisCache("cache", + RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory, BatchStrategy.scan(25)), + RedisCacheConfiguration.defaultCacheConfig().serializeValuesWith(SerializationPair.fromSerializer(serializer))); + + doWithConnection(connection -> { + connection.set(binaryCacheKey, binaryNullValue); + connection.set("cache::foo".getBytes(), binaryNullValue); + connection.set("other".getBytes(), "value".getBytes()); + }); + + cache.clear(); + + doWithConnection(connection -> { + assertThat(connection.exists(binaryCacheKey)).isFalse(); + assertThat(connection.exists("cache::foo".getBytes())).isFalse(); + assertThat(connection.exists("other".getBytes())).isTrue(); + }); + } + @ParameterizedRedisTest // DATAREDIS-481 void getWithCallableShouldResolveValueIfNotPresent() { @@ -280,7 +309,8 @@ public class RedisCacheTests { @ParameterizedRedisTest // DATAREDIS-715 void computePrefixCreatesCacheKeyCorrectly() { - RedisCache cacheWithCustomPrefix = new RedisCache("cache", new DefaultRedisCacheWriter(connectionFactory), + RedisCache cacheWithCustomPrefix = new RedisCache("cache", + RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory), RedisCacheConfiguration.defaultCacheConfig().serializeValuesWith(SerializationPair.fromSerializer(serializer)) .computePrefixWith(cacheName -> "_" + cacheName + "_")); @@ -296,7 +326,8 @@ public class RedisCacheTests { @ParameterizedRedisTest // DATAREDIS-1041 void prefixCacheNameCreatesCacheKeyCorrectly() { - RedisCache cacheWithCustomPrefix = new RedisCache("cache", new DefaultRedisCacheWriter(connectionFactory), + RedisCache cacheWithCustomPrefix = new RedisCache("cache", + RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory), RedisCacheConfiguration.defaultCacheConfig().serializeValuesWith(SerializationPair.fromSerializer(serializer)) .prefixCacheNameWith("redis::")); @@ -314,7 +345,8 @@ public class RedisCacheTests { doWithConnection(connection -> connection.set("_cache_key-1".getBytes(StandardCharsets.UTF_8), binarySample)); - RedisCache cacheWithCustomPrefix = new RedisCache("cache", new DefaultRedisCacheWriter(connectionFactory), + RedisCache cacheWithCustomPrefix = new RedisCache("cache", + RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory), RedisCacheConfiguration.defaultCacheConfig().serializeValuesWith(SerializationPair.fromSerializer(serializer)) .computePrefixWith(cacheName -> "_" + cacheName + "_"));