Transition BatchStrategy to interface.
...and use long return type instead of int for cleanCache. Original Pull Request: #2051
This commit is contained in:
Christoph Strobl
162
src/main/java/org/springframework/data/redis/cache/BatchStrategies.java
vendored
Normal file
162
src/main/java/org/springframework/data/redis/cache/BatchStrategies.java
vendored
Normal file
@@ -0,0 +1,162 @@
|
||||
/*
|
||||
* 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;
|
||||
|
||||
/**
|
||||
* A collection of predefined {@link BatchStrategy} implementations using {@code KEYS} or {@code SCAN} command.
|
||||
*
|
||||
* @author Mark Paluch
|
||||
* @author Christoph Strobl
|
||||
* @since 2.6
|
||||
*/
|
||||
public abstract class BatchStrategies {
|
||||
|
||||
private BatchStrategies() {
|
||||
// can't touch this - oh-oh oh oh oh-oh-oh
|
||||
}
|
||||
|
||||
/**
|
||||
* A {@link BatchStrategy} 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
|
||||
* on large keyspaces.
|
||||
* <p/>
|
||||
* {@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;
|
||||
}
|
||||
|
||||
/**
|
||||
* A {@link BatchStrategy} 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.
|
||||
* <p/>
|
||||
* 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);
|
||||
}
|
||||
|
||||
/**
|
||||
* {@link BatchStrategy} using {@code KEYS}.
|
||||
*/
|
||||
static class Keys implements BatchStrategy {
|
||||
|
||||
static Keys INSTANCE = new Keys();
|
||||
|
||||
@Override
|
||||
public long 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 implements BatchStrategy {
|
||||
|
||||
private final int batchSize;
|
||||
|
||||
Scan(int batchSize) {
|
||||
this.batchSize = batchSize;
|
||||
}
|
||||
|
||||
@Override
|
||||
public long cleanCache(RedisConnection connection, String name, byte[] pattern) {
|
||||
|
||||
Cursor<byte[]> cursor = connection.scan(ScanOptions.scanOptions().count(batchSize).match(pattern).build());
|
||||
|
||||
long count = 0;
|
||||
|
||||
PartitionIterator<byte[]> partitions = new PartitionIterator<>(cursor, batchSize);
|
||||
while (partitions.hasNext()) {
|
||||
|
||||
List<byte[]> 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 <T>
|
||||
*/
|
||||
static class PartitionIterator<T> implements Iterator<List<T>> {
|
||||
|
||||
private final Iterator<T> iterator;
|
||||
private final int size;
|
||||
|
||||
PartitionIterator(Iterator<T> iterator, int size) {
|
||||
|
||||
this.iterator = iterator;
|
||||
this.size = size;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean hasNext() {
|
||||
return iterator.hasNext();
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<T> next() {
|
||||
|
||||
if (!hasNext()) {
|
||||
throw new NoSuchElementException();
|
||||
}
|
||||
|
||||
List<T> list = new ArrayList<>(size);
|
||||
while (list.size() < size && iterator.hasNext()) {
|
||||
list.add(iterator.next());
|
||||
}
|
||||
|
||||
return list;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -15,156 +15,30 @@
|
||||
*/
|
||||
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}.
|
||||
* A {@link BatchStrategy} to be used with {@link RedisCacheWriter}.
|
||||
* <p/>
|
||||
* Primarily used to clear the cache.
|
||||
* Mainly used to clear the cache.
|
||||
* <p/>
|
||||
* Predefined strategies using the {@link BatchStrategies#keys() KEYS} or {@link BatchStrategies#scan(int) SCAN}
|
||||
* commands can be found in {@link BatchStrategies}.
|
||||
*
|
||||
* @author Mark Paluch
|
||||
* @author Christoph Strobl
|
||||
* @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 on large
|
||||
* keyspaces.
|
||||
* <p/>
|
||||
* {@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.
|
||||
* <p/>
|
||||
* 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);
|
||||
}
|
||||
public interface BatchStrategy {
|
||||
|
||||
/**
|
||||
* Remove all keys following the given pattern.
|
||||
*
|
||||
* @param the connection to use.
|
||||
* @param name The cache name must not be {@literal null}.
|
||||
* @param connection the connection to use. Must not be {@literal null}.
|
||||
* @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<byte[]> cursor = connection.scan(ScanOptions.scanOptions().count(batchSize).match(pattern).build());
|
||||
|
||||
PartitionIterator<byte[]> partitions = new PartitionIterator<>(cursor, batchSize);
|
||||
|
||||
int count = 0;
|
||||
|
||||
while (partitions.hasNext()) {
|
||||
|
||||
List<byte[]> 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 <T>
|
||||
*/
|
||||
static class PartitionIterator<T> implements Iterator<List<T>> {
|
||||
|
||||
private final Iterator<T> iterator;
|
||||
private final int size;
|
||||
|
||||
public PartitionIterator(Iterator<T> iterator, int size) {
|
||||
this.iterator = iterator;
|
||||
this.size = size;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean hasNext() {
|
||||
return iterator.hasNext();
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<T> next() {
|
||||
|
||||
if (!hasNext()) {
|
||||
throw new NoSuchElementException();
|
||||
}
|
||||
|
||||
List<T> list = new ArrayList<>(size);
|
||||
while (list.size() < size && iterator.hasNext()) {
|
||||
list.add(iterator.next());
|
||||
}
|
||||
|
||||
return list;
|
||||
}
|
||||
}
|
||||
long cleanCache(RedisConnection connection, String name, byte[] pattern);
|
||||
|
||||
}
|
||||
|
||||
@@ -217,8 +217,12 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
|
||||
wasLocked = true;
|
||||
}
|
||||
|
||||
|
||||
statistics.incDeletesBy(name, batchStrategy.cleanCache(connection, name, pattern));
|
||||
long deleteCount = batchStrategy.cleanCache(connection, name, pattern);
|
||||
while (deleteCount > Integer.MAX_VALUE) {
|
||||
statistics.incDeletesBy(name, Integer.MAX_VALUE);
|
||||
deleteCount -= Integer.MAX_VALUE;
|
||||
}
|
||||
statistics.incDeletesBy(name, (int) deleteCount);
|
||||
|
||||
} finally {
|
||||
|
||||
|
||||
@@ -43,7 +43,7 @@ public interface RedisCacheWriter extends CacheStatisticsProvider {
|
||||
* @return new instance of {@link DefaultRedisCacheWriter}.
|
||||
*/
|
||||
static RedisCacheWriter nonLockingRedisCacheWriter(RedisConnectionFactory connectionFactory) {
|
||||
return nonLockingRedisCacheWriter(connectionFactory, BatchStrategy.keys());
|
||||
return nonLockingRedisCacheWriter(connectionFactory, BatchStrategies.keys());
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -70,7 +70,7 @@ public interface RedisCacheWriter extends CacheStatisticsProvider {
|
||||
* @return new instance of {@link DefaultRedisCacheWriter}.
|
||||
*/
|
||||
static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectionFactory) {
|
||||
return lockingRedisCacheWriter(connectionFactory, BatchStrategy.keys());
|
||||
return lockingRedisCacheWriter(connectionFactory, BatchStrategies.keys());
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user