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

DATAREDIS-635 - Add keyspace scanning using Redis Cluster.

Browse Source 
We now support SCAN commands with Redis Cluster. Cluster-wide scanning is supported with Lettuce only and SCAN on a selected node works with both clients.

// Lettuce only
Cursor<byte[]> scan = clusterConnection.scan(ScanOptions.NONE);

// Jedis and Lettuce
Cursor<byte[]> scan = clusterConnection.scan(new RedisClusterNode("127.0.0.1", 7379, SlotRange.empty()), ScanOptions.NONE);

Original Pull Request: #296
This commit is contained in:
Mark Paluch
2017-12-14 15:36:04 +01:00
committed by Christoph Strobl
parent 871465fa3b
commit c18a311aa9
12 changed files with 296 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
src/main/java/org/springframework/data/redis/connection/RedisClusterConnection.java
View File

@@ -18,6 +18,8 @@ package org.springframework.data.redis.connection;
import java.util.Collection;
import java.util.Set;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.lang.Nullable;
/**
@@ -49,6 +51,17 @@ public interface RedisClusterConnection extends RedisConnection, RedisClusterCom
@Nullable
Set<byte[]> keys(RedisClusterNode node, byte[] pattern);
/**
* Use a {@link Cursor} to iterate over keys.
*
* @param node must not be {@literal null}.
* @param options must not be {@literal null}.
* @return never {@literal null}.
* @since 2.1
* @see <a href="http://redis.io/commands/scan">Redis Documentation: SCAN</a>
*/
Cursor<byte[]> scan(RedisClusterNode node, ScanOptions options);
/**
* @param node must not be {@literal null}.
* @return {@literal null} when no keys stored at node or when used in pipeline / transaction.

15
src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java
View File

@@ -48,6 +48,8 @@ import org.springframework.data.redis.connection.ClusterCommandExecutor.MultiKey
import org.springframework.data.redis.connection.ClusterCommandExecutor.NodeResult;
import org.springframework.data.redis.connection.RedisClusterNode.SlotRange;
import org.springframework.data.redis.connection.convert.Converters;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.util.DirectFieldAccessFallbackBeanWrapper;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
@@ -304,11 +306,24 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
return new JedisClusterKeyCommands(this);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisClusterConnection#keys(org.springframework.data.redis.connection.RedisClusterNode, byte[])
*/
@Override
public Set<byte[]> keys(RedisClusterNode node, byte[] pattern) {
return doGetKeyCommands().keys(node, pattern);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisClusterConnection#scan(org.springframework.data.redis.connection.RedisClusterNode, org.springframework.data.redis.core.ScanOptions)
*/
@Override
public Cursor<byte[]> scan(RedisClusterNode node, ScanOptions options) {
return doGetKeyCommands().scan(node, options);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisClusterConnection#randomKey(org.springframework.data.redis.connection.RedisClusterNode)

25
src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterKeyCommands.java
View File

@@ -18,6 +18,7 @@ package org.springframework.data.redis.connection.jedis;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import redis.clients.jedis.BinaryJedis;
import redis.clients.jedis.ScanParams;
import java.util.ArrayList;
import java.util.Arrays;
@@ -41,6 +42,8 @@ import org.springframework.data.redis.connection.convert.Converters;
import org.springframework.data.redis.connection.jedis.JedisClusterConnection.JedisClusterCommandCallback;
import org.springframework.data.redis.connection.jedis.JedisClusterConnection.JedisMultiKeyClusterCommandCallback;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanCursor;
import org.springframework.data.redis.core.ScanIteration;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
@@ -168,6 +171,28 @@ class JedisClusterKeyCommands implements RedisKeyCommands {
throw new InvalidDataAccessApiUsageException("Scan is not supported across multiple nodes within a cluster");
}
Cursor<byte[]> scan(RedisClusterNode node, ScanOptions options) {
Assert.notNull(node, "RedisClusterNode must not be null!");
Assert.notNull(options, "Pattern must not be null!");
return connection.getClusterCommandExecutor()
.executeCommandOnSingleNode((JedisClusterCommandCallback<Cursor<byte[]>>) client -> {
return new ScanCursor<byte[]>(0, options) {
@Override
protected ScanIteration<byte[]> doScan(long cursorId, ScanOptions options) {
ScanParams params = JedisConverters.toScanParams(options);
redis.clients.jedis.ScanResult<String> result = client.scan(Long.toString(cursorId), params);
return new ScanIteration<>(Long.valueOf(result.getStringCursor()),
JedisConverters.stringListToByteList().convert(result.getResult()));
}
}.open();
}, node).getValue();
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisKeyCommands#randomKey()

15
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java
View File

@@ -46,6 +46,8 @@ import org.springframework.data.redis.connection.ClusterCommandExecutor.MultiKey
import org.springframework.data.redis.connection.ClusterCommandExecutor.NodeResult;
import org.springframework.data.redis.connection.RedisClusterNode.SlotRange;
import org.springframework.data.redis.connection.convert.Converters;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
@@ -491,6 +493,19 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
return doGetClusterKeyCommands().keys(node, pattern);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisClusterConnection#scan(org.springframework.data.redis.connection.RedisClusterNode, org.springframework.data.redis.core.ScanOptions)
*/
@Override
public Cursor<byte[]> scan(RedisClusterNode node, ScanOptions options) {
return doGetClusterKeyCommands().scan(node, options);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisClusterConnection#randomKey(org.springframework.data.redis.connection.RedisClusterNode)
*/
public byte[] randomKey(RedisClusterNode node) {
return doGetClusterKeyCommands().randomKey(node);
}

47
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterKeyCommands.java
View File

@@ -15,18 +15,21 @@
*/
package org.springframework.data.redis.connection.lettuce;
import io.lettuce.core.KeyScanCursor;
import io.lettuce.core.ScanArgs;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.redis.connection.ClusterSlotHashUtil;
import org.springframework.data.redis.connection.RedisClusterNode;
import org.springframework.data.redis.connection.SortParameters;
import org.springframework.data.redis.connection.lettuce.LettuceClusterConnection.LettuceClusterCommandCallback;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanCursor;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
@@ -47,15 +50,6 @@ class LettuceClusterKeyCommands extends LettuceKeyCommands {
this.connection = connection;
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceConnection#scan(long, org.springframework.data.redis.core.ScanOptions)
*/
@Override
public Cursor<byte[]> scan(long cursorId, ScanOptions options) {
throw new InvalidDataAccessApiUsageException("Scan is not supported across multiple nodes within a cluster.");
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceConnection#randomKey()
@@ -155,8 +149,8 @@ class LettuceClusterKeyCommands extends LettuceKeyCommands {
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceConnection#move(byte[], int)
* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceConnection#move(byte[], int)
*/
@Override
public Boolean move(byte[] key, int dbIndex) {
@@ -164,8 +158,8 @@ class LettuceClusterKeyCommands extends LettuceKeyCommands {
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisClusterConnection#randomKey(org.springframework.data.redis.connection.RedisClusterNode)
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisClusterConnection#randomKey(org.springframework.data.redis.connection.RedisClusterNode)
*/
@Nullable
public byte[] randomKey(RedisClusterNode node) {
@@ -189,6 +183,31 @@ class LettuceClusterKeyCommands extends LettuceKeyCommands {
.getValue());
}
Cursor<byte[]> scan(RedisClusterNode node, ScanOptions options) {
Assert.notNull(node, "RedisClusterNode must not be null!");
Assert.notNull(options, "Pattern must not be null!");
return connection.getClusterCommandExecutor()
.executeCommandOnSingleNode((LettuceClusterCommandCallback<ScanCursor<byte[]>>) client -> {
return new LettuceScanCursor<byte[]>(options) {
@Override
protected LettuceScanIteration<byte[]> doScan(io.lettuce.core.ScanCursor cursor, ScanOptions options) {
ScanArgs scanArgs = connection.getScanArgs(options);
KeyScanCursor<byte[]> keyScanCursor = client.scan(cursor, scanArgs);
List<byte[]> keys = keyScanCursor.getKeys();
return new LettuceScanIteration<>(keyScanCursor, keys);
}
}.open();
}, node).getValue();
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceConnection#sort(byte[], org.springframework.data.redis.connection.SortParameters, byte[])

22
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceKeyCommands.java
View File

@@ -17,6 +17,7 @@ package org.springframework.data.redis.connection.lettuce;
import io.lettuce.core.KeyScanCursor;
import io.lettuce.core.ScanArgs;
import io.lettuce.core.ScanCursor;
import io.lettuce.core.SortArgs;
import io.lettuce.core.cluster.api.async.RedisClusterAsyncCommands;
import io.lettuce.core.cluster.api.sync.RedisClusterCommands;
@@ -33,8 +34,6 @@ import org.springframework.data.redis.connection.RedisKeyCommands;
import org.springframework.data.redis.connection.SortParameters;
import org.springframework.data.redis.connection.convert.Converters;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanCursor;
import org.springframework.data.redis.core.ScanIteration;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
@@ -229,7 +228,7 @@ class LettuceKeyCommands implements RedisKeyCommands {
* @return
*/
public Cursor<byte[]> scan() {
return scan(0, ScanOptions.NONE);
return scan(ScanOptions.NONE);
}
/*
@@ -238,36 +237,31 @@ class LettuceKeyCommands implements RedisKeyCommands {
*/
@Override
public Cursor<byte[]> scan(ScanOptions options) {
return scan(0, options != null ? options : ScanOptions.NONE);
return doScan(options != null ? options : ScanOptions.NONE);
}
/**
* @since 1.4
* @param cursorId
* @param options
* @return
*/
public Cursor<byte[]> scan(long cursorId, ScanOptions options) {
private Cursor<byte[]> doScan(ScanOptions options) {
return new ScanCursor<byte[]>(cursorId, options) {
return new LettuceScanCursor<byte[]>(options) {
@SuppressWarnings("unchecked")
@Override
protected ScanIteration<byte[]> doScan(long cursorId, ScanOptions options) {
protected LettuceScanIteration<byte[]> doScan(ScanCursor cursor, ScanOptions options) {
if (isQueueing() || isPipelined()) {
throw new UnsupportedOperationException("'SCAN' cannot be called in pipeline / transaction mode.");
}
io.lettuce.core.ScanCursor scanCursor = connection.getScanCursor(cursorId);
ScanArgs scanArgs = connection.getScanArgs(options);
KeyScanCursor<byte[]> keyScanCursor = getConnection().scan(scanCursor, scanArgs);
String nextCursorId = keyScanCursor.getCursor();
KeyScanCursor<byte[]> keyScanCursor = getConnection().scan(cursor, scanArgs);
List<byte[]> keys = keyScanCursor.getKeys();
return new ScanIteration<>(Long.valueOf(nextCursorId), (keys));
return new LettuceScanIteration<>(keyScanCursor, keys);
}
@Override

113
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceScanCursor.java Normal file
View File

@@ -0,0 +1,113 @@
/*
* Copyright 2017 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
*
* http://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.connection.lettuce;
import java.util.Collection;
import org.springframework.data.redis.core.ScanCursor;
import org.springframework.data.redis.core.ScanIteration;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.lang.Nullable;
/**
* Lettuce-specific {@link ScanCursor} extension that maintains the cursor state that is required for stateful-scanning
* across a Redis Cluster.
* <p/>
* The cursor state uses Lettuce's stateful {@link io.lettuce.core.ScanCursor} to keep track of scanning progress.
* Lettuce's cursor stores scanning progress inside its cursor so using a primitive {@code long} is not sufficient.
*
* @author Mark Paluch
* @since 2.1
*/
abstract class LettuceScanCursor<T> extends ScanCursor<T> {
@Nullable private io.lettuce.core.ScanCursor state;
/**
* Creates a new {@link LettuceScanCursor} given {@link ScanOptions}.
*
* @param options must not be {@literal null}.
*/
LettuceScanCursor(ScanOptions options) {
super(options);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.core.ScanCursor#doScan(long, org.springframework.data.redis.core.ScanOptions)
*/
@Override
protected ScanIteration<T> doScan(long cursorId, ScanOptions options) {
if (state == null && cursorId == 0) {
return scanAndProcessState(io.lettuce.core.ScanCursor.INITIAL, options);
}
if (state != null) {
if (isMatchingCursor(cursorId)) {
return scanAndProcessState(state, options);
}
}
throw new IllegalArgumentException(String.format("Current scan %s state and cursor %d do not match!",
state != null ? state.getCursor() : "(none)", cursorId));
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.core.ScanCursor#isFinished(long)
*/
@Override
protected boolean isFinished(long cursorId) {
return state != null && isMatchingCursor(cursorId) ? state.isFinished() : super.isFinished(cursorId);
}
private ScanIteration<T> scanAndProcessState(io.lettuce.core.ScanCursor scanCursor, ScanOptions options) {
LettuceScanIteration<T> iteration = doScan(scanCursor, options);
state = iteration.cursor;
return iteration;
}
private boolean isMatchingCursor(long cursorId) {
return state != null && state.getCursor().equals(Long.toString(cursorId));
}
/**
* Perform the actual scan operation given {@link io.lettuce.core.ScanCursor} and {@link ScanOptions}.
*
* @param cursor must not be {@literal null}.
* @param options must not be {@literal null}.
* @return
*/
protected abstract LettuceScanIteration<T> doScan(io.lettuce.core.ScanCursor cursor, ScanOptions options);
/**
* Lettuce-specific extension to {@link ScanIteration} keeping track of the original
* {@link io.lettuce.core.ScanCursor} object.
*/
static class LettuceScanIteration<T> extends ScanIteration<T> {
private final io.lettuce.core.ScanCursor cursor;
LettuceScanIteration(io.lettuce.core.ScanCursor cursor, Collection<T> items) {
super(Long.parseLong(cursor.getCursor()), items);
this.cursor = cursor;
}
}
}

13
src/main/java/org/springframework/data/redis/core/ScanCursor.java
View File

@@ -134,7 +134,7 @@ public abstract class ScanCursor<T> implements Cursor<T> {
cursorId = Long.valueOf(result.getCursorId());
if (cursorId == 0) {
if (isFinished(cursorId)) {
state = CursorState.FINISHED;
}
@@ -145,6 +145,17 @@ public abstract class ScanCursor<T> implements Cursor<T> {
}
}
/**
* Check whether {@code cursorId} is finished.
*
* @param cursorId the cursor Id
* @return {@literal true} if the cursor is considered finished, {@literal false} otherwise.s
* @since 2.1
*/
protected boolean isFinished(long cursorId) {
return cursorId == 0;
}
private void resetDelegate() {
delegate = Collections.<T> emptyList().iterator();
}