From 7ec4641cc1e2bec7250bf426b22c05d4369411db Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Thu, 13 Dec 2018 11:23:22 +0100 Subject: [PATCH] DATAREDIS-913 - Delete multiple keys through ReactiveRedisTemplate.delete(Publisher) and unlink(Publisher). 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 --- .../redis/core/ReactiveRedisOperations.java | 6 ++- .../redis/core/ReactiveRedisTemplate.java | 16 ++++---- ...ReactiveRedisTemplateIntegrationTests.java | 39 +++++++++++++++++++ 3 files changed, 52 insertions(+), 9 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java index ae5a9bd8e..a630ed399 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java @@ -213,7 +213,8 @@ public interface ReactiveRedisOperations { Mono 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 { /** * 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. diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java index 07ec87e0b..3be8e054f 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java @@ -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 implements ReactiveRedisOperations 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 implements ReactiveRedisOperations 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)); } /* diff --git a/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java index 2a03d312e..3590a3072 100644 --- a/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java @@ -18,6 +18,7 @@ package org.springframework.data.redis.core; import static org.assertj.core.api.Assertions.*; import static org.junit.Assume.*; +import org.springframework.data.redis.StringObjectFactory; import reactor.core.publisher.Flux; import reactor.test.StepVerifier; @@ -217,6 +218,44 @@ public class ReactiveRedisTemplateIntegrationTests { StepVerifier.create(redisTemplate.hasKey(key1)).expectNext(false).verifyComplete(); StepVerifier.create(redisTemplate.hasKey(key2)).expectNext(false).verifyComplete(); } + + @Test // DATAREDIS-913 + public void unlinkManyPublisher() { + + K key1 = keyFactory.instance(); + K key2 = keyFactory.instance(); + + assumeTrue(key1 instanceof String && valueFactory instanceof StringObjectFactory); + + StepVerifier.create(redisTemplate.opsForValue().set(key1, valueFactory.instance())).expectNext(true) + .verifyComplete(); + StepVerifier.create(redisTemplate.opsForValue().set(key2, valueFactory.instance())).expectNext(true) + .verifyComplete(); + + StepVerifier.create(redisTemplate.unlink(redisTemplate.keys((K) "*"))).expectNext(2L).verifyComplete(); + + StepVerifier.create(redisTemplate.hasKey(key1)).expectNext(false).verifyComplete(); + StepVerifier.create(redisTemplate.hasKey(key2)).expectNext(false).verifyComplete(); + } + + @Test // DATAREDIS-913 + public void deleteManyPublisher() { + + K key1 = keyFactory.instance(); + K key2 = keyFactory.instance(); + + assumeTrue(key1 instanceof String && valueFactory instanceof StringObjectFactory); + + StepVerifier.create(redisTemplate.opsForValue().set(key1, valueFactory.instance())).expectNext(true) + .verifyComplete(); + StepVerifier.create(redisTemplate.opsForValue().set(key2, valueFactory.instance())).expectNext(true) + .verifyComplete(); + + StepVerifier.create(redisTemplate.delete(redisTemplate.keys((K) "*"))).expectNext(2L).verifyComplete(); + + StepVerifier.create(redisTemplate.hasKey(key1)).expectNext(false).verifyComplete(); + StepVerifier.create(redisTemplate.hasKey(key2)).expectNext(false).verifyComplete(); + } @Test // DATAREDIS-683 @SuppressWarnings("unchecked")