Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAREDIS-913 - Delete multiple keys through ReactiveRedisTemplate.delete(Publisher) and unlink(Publisher).

Browse Source 
We now delete all keys that are emitted by the source publisher when invoking delete(Publisher) and unlink(Publisher). Both methods split incoming keys into chunks of 128 keys to keep a balance between latency and the issued command count.

Previously, delete(Publisher) and unlink(Publisher) only deleted the first key.

Original Pull Request: #379
This commit is contained in:
Mark Paluch
2018-12-13 11:23:22 +01:00
committed by Christoph Strobl
parent d248c0cd7c
commit 7ec4641cc1
3 changed files with 52 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java
View File

@@ -213,7 +213,8 @@ public interface ReactiveRedisOperations<K, V> {
Mono<Long> delete(K... key);
/**
* Delete given {@code keys}.
* Delete given {@code keys}. This command buffers keys received from {@link Publisher} into chunks of 128 keys to
* delete to reduce the number of issued {@code DEL} commands.
*
* @param keys must not be {@literal null}.
* @return The number of keys that were removed.
@@ -234,7 +235,8 @@ public interface ReactiveRedisOperations<K, V> {
/**
* Unlink the {@code keys} from the keyspace. Unlike with {@link #delete(Publisher)} the actual memory reclaiming here
* happens asynchronously.
* happens asynchronously. This command buffers keys received from {@link Publisher} into chunks of 128 keys to delete
* to reduce the number of issued {@code UNLINK} commands.
*
* @param keys must not be {@literal null}.
* @return The number of keys that were removed. {@literal null} when used in pipeline / transaction.

16
src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java
View File

@@ -28,12 +28,12 @@ import java.time.Duration;
import java.time.Instant;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
import org.reactivestreams.Publisher;
import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.connection.ReactiveRedisConnection;
import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
import org.springframework.data.redis.connection.ReactiveSubscription.Message;
import org.springframework.data.redis.core.script.DefaultReactiveScriptExecutor;
@@ -350,9 +350,10 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
Assert.notNull(keys, "Keys must not be null!");
return createMono(connection -> connection.keyCommands() //
.del(Flux.from(keys).map(this::rawKey).map(KeyCommand::new)) //
.map(CommandResponse::getOutput));
return createFlux(connection -> connection.keyCommands() //
.mDel(Flux.from(keys).map(this::rawKey).buffer(128)) //
.map(CommandResponse::getOutput)) //
.collect(Collectors.summingLong(value -> value));
}
/*
@@ -384,9 +385,10 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
Assert.notNull(keys, "Keys must not be null!");
return createMono(connection -> connection.keyCommands() //
.unlink(Flux.from(keys).map(this::rawKey).map(KeyCommand::new)) //
.map(CommandResponse::getOutput));
return createFlux(connection -> connection.keyCommands() //
.mUnlink(Flux.from(keys).map(this::rawKey).buffer(128)) //
.map(CommandResponse::getOutput)) //
.collect(Collectors.summingLong(value -> value));
}
/*