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

DATAREDIS-1011 - Allow configuration of Lettuce pipelining flush behavior.

Browse Source 
We now allow customization of flushing when using pipelining with Lettuce. Lettuce flushes each command by default. To optimize for performance, flushing behavior can be customized using a PipeliningFlushPolicy. Flush each command, flush on close and flush after n commands are the bundled implementations that can be configured on LettuceConnectionFactory.

Original Pull Request: #511
This commit is contained in:
Mark Paluch
2020-02-13 16:25:23 +01:00
committed by Christoph Strobl
parent a918d83526
commit 42ad8d190d
6 changed files with 445 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

268
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java
View File

@@ -53,6 +53,7 @@ import java.util.concurrent.CompletionStage;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Supplier;
import org.springframework.beans.BeanUtils;
@@ -64,9 +65,9 @@ import org.springframework.data.redis.ExceptionTranslationStrategy;
import org.springframework.data.redis.FallbackExceptionTranslationStrategy;
import org.springframework.data.redis.connection.*;
import org.springframework.data.redis.connection.convert.TransactionResultConverter;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware;
import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceResultBuilder;
import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceStatusResult;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware;
import org.springframework.data.redis.core.RedisCommand;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
@@ -108,10 +109,12 @@ public class LettuceConnection extends AbstractRedisConnection {
private boolean isMulti = false;
private boolean isPipelined = false;
private @Nullable List<LettuceResult> ppline;
private @Nullable PipeliningFlushState flushState;
private final Queue<FutureResult<?>> txResults = new LinkedList<>();
private volatile @Nullable LettuceSubscription subscription;
/** flag indicating whether the connection needs to be dropped or not */
private boolean convertPipelineAndTxResults = true;
private PipeliningFlushPolicy pipeliningFlushPolicy = PipeliningFlushPolicy.flushEachCommand();
LettuceResult newLettuceResult(Future<?> resultHolder) {
return newLettuceResult(resultHolder, (val) -> val);
@@ -518,6 +521,8 @@ public class LettuceConnection extends AbstractRedisConnection {
if (!isPipelined) {
isPipelined = true;
ppline = new ArrayList<>();
flushState = this.pipeliningFlushPolicy.newPipeline();
flushState.onOpen(this.getOrCreateDedicatedConnection());
}
}
@@ -529,9 +534,11 @@ public class LettuceConnection extends AbstractRedisConnection {
public List<Object> closePipeline() {
if (!isPipelined) {
return Collections.emptyList();
}
flushState.onClose(this.getOrCreateDedicatedConnection());
flushState = null;
isPipelined = false;
List<io.lettuce.core.protocol.RedisCommand<?, ?, ?>> futures = new ArrayList<>(ppline.size());
for (LettuceResult<?, ?> result : ppline) {
@@ -863,6 +870,21 @@ public class LettuceConnection extends AbstractRedisConnection {
this.convertPipelineAndTxResults = convertPipelineAndTxResults;
}
/**
* Configures the flushing policy when using pipelining.
*
* @param pipeliningFlushPolicy the flushing policy to control when commands get written to the Redis connection.
* @see PipeliningFlushPolicy#flushEachCommand()
* @see #openPipeline()
* @see StatefulRedisConnection#flushCommands()
*/
public void setPipeliningFlushPolicy(PipeliningFlushPolicy pipeliningFlushPolicy) {
Assert.notNull(pipeliningFlushPolicy, "PipeliningFlushingPolicy must not be null!");
this.pipeliningFlushPolicy = pipeliningFlushPolicy;
}
private void checkSubscription() {
if (isSubscribed()) {
throw new RedisSubscribedConnectionException(
@@ -902,8 +924,14 @@ public class LettuceConnection extends AbstractRedisConnection {
void pipeline(LettuceResult result) {
if (isQueueing()) {
if (flushState != null) {
flushState.onCommand(getOrCreateDedicatedConnection());
}
transaction(result);
} else {
flushState.onCommand(getOrCreateDedicatedConnection());
ppline.add(result);
}
}
@@ -947,31 +975,43 @@ public class LettuceConnection extends AbstractRedisConnection {
return (RedisAsyncCommands) getAsyncDedicatedConnection();
}
RedisClusterCommands<byte[], byte[]> getDedicatedConnection() {
StatefulConnection<byte[], byte[]> connection = getOrCreateDedicatedConnection();
if (connection instanceof StatefulRedisConnection) {
return ((StatefulRedisConnection<byte[], byte[]>) connection).sync();
}
if (connection instanceof StatefulRedisClusterConnection) {
return ((StatefulRedisClusterConnection<byte[], byte[]>) connection).sync();
}
throw new IllegalStateException(
String.format("%s is not a supported connection type.", connection.getClass().getName()));
}
protected RedisClusterAsyncCommands<byte[], byte[]> getAsyncDedicatedConnection() {
StatefulConnection<byte[], byte[]> connection = getOrCreateDedicatedConnection();
if (connection instanceof StatefulRedisConnection) {
return ((StatefulRedisConnection<byte[], byte[]>) connection).async();
}
if (asyncDedicatedConn instanceof StatefulRedisClusterConnection) {
return ((StatefulRedisClusterConnection<byte[], byte[]>) connection).async();
}
throw new IllegalStateException(
String.format("%s is not a supported connection type.", connection.getClass().getName()));
}
private StatefulConnection<byte[], byte[]> getOrCreateDedicatedConnection() {
if (asyncDedicatedConn == null) {
asyncDedicatedConn = doGetAsyncDedicatedConnection();
}
if (asyncDedicatedConn instanceof StatefulRedisConnection) {
return ((StatefulRedisConnection<byte[], byte[]>) asyncDedicatedConn).async();
}
if (asyncDedicatedConn instanceof StatefulRedisClusterConnection) {
return ((StatefulRedisClusterConnection<byte[], byte[]>) asyncDedicatedConn).async();
}
throw new IllegalStateException(
String.format("%s is not a supported connection type.", asyncDedicatedConn.getClass().getName()));
}
private boolean customizedDatabaseIndex() {
return defaultDbIndex != dbIndex;
}
private void potentiallySelectDatabase(int dbIndex) {
if (asyncDedicatedConn instanceof StatefulRedisConnection) {
((StatefulRedisConnection<byte[], byte[]>) asyncDedicatedConn).sync().select(dbIndex);
}
return asyncDedicatedConn;
}
@SuppressWarnings("unchecked")
@@ -979,23 +1019,6 @@ public class LettuceConnection extends AbstractRedisConnection {
return (RedisCommands) getDedicatedConnection();
}
RedisClusterCommands<byte[], byte[]> getDedicatedConnection() {
if (asyncDedicatedConn == null) {
asyncDedicatedConn = doGetAsyncDedicatedConnection();
}
if (asyncDedicatedConn instanceof StatefulRedisConnection) {
return ((StatefulRedisConnection<byte[], byte[]>) asyncDedicatedConn).sync();
}
if (asyncDedicatedConn instanceof StatefulRedisClusterConnection) {
return ((StatefulRedisClusterConnection<byte[], byte[]>) asyncDedicatedConn).sync();
}
throw new IllegalStateException(
String.format("%s is not a supported connection type.", asyncDedicatedConn.getClass().getName()));
}
@SuppressWarnings("unchecked")
protected StatefulConnection<byte[], byte[]> doGetAsyncDedicatedConnection() {
@@ -1008,6 +1031,16 @@ public class LettuceConnection extends AbstractRedisConnection {
return connection;
}
private boolean customizedDatabaseIndex() {
return defaultDbIndex != dbIndex;
}
private void potentiallySelectDatabase(int dbIndex) {
if (asyncDedicatedConn instanceof StatefulRedisConnection) {
((StatefulRedisConnection<byte[], byte[]>) asyncDedicatedConn).sync().select(dbIndex);
}
}
io.lettuce.core.ScanCursor getScanCursor(long cursorId) {
return io.lettuce.core.ScanCursor.of(Long.toString(cursorId));
}
@@ -1319,4 +1352,163 @@ public class LettuceConnection extends AbstractRedisConnection {
}
}
}
/**
* Strategy interface to control pipelining flush behavior. Lettuce writes (flushes) each command individually to the
* Redis connection. Flushing behavior can be customized to optimize for performance. Flushing can be either stateless
* or stateful. An example for stateful flushing is size-based (buffer) flushing to flush after a configured number of
* commands.
*
* @see StatefulRedisConnection#setAutoFlushCommands(boolean)
* @see StatefulRedisConnection#flushCommands()
*/
public interface PipeliningFlushPolicy {
/**
* Return a policy to flush after each command (default behavior).
*
* @return a policy to flush after each command.
*/
static PipeliningFlushPolicy flushEachCommand() {
return FlushEachCommand.INSTANCE;
}
/**
* Return a policy to flush only if {@link #closePipeline()} is called.
*
* @return a policy to flush after each command.
*/
static PipeliningFlushPolicy flushOnClose() {
return FlushOnClose.INSTANCE;
}
/**
* Return a policy to buffer commands and to flush once reaching the configured {@code bufferSize}. The buffer is
* recurring so a buffer size of e.g. {@code 2} will flush after 2, 4, 6, … commands.
*
* @param bufferSize the number of commands to buffer before flushing. Must be greater than zero.
* @return a policy to flush buffered commands to the Redis connection once the configured number of commands was
* issued.
*/
static PipeliningFlushPolicy buffered(int bufferSize) {
Assert.isTrue(bufferSize > 0, "Buffer size must be greater than 0");
return () -> new BufferedFlushing(bufferSize);
}
PipeliningFlushState newPipeline();
}
/**
* State object associated with flushing of the currently ongoing pipeline.
*/
public interface PipeliningFlushState {
/**
* Callback if the pipeline gets opened.
*
* @param connection
* @see #openPipeline()
*/
void onOpen(StatefulConnection<?, ?> connection);
/**
* Callback for each issued Redis command.
*
* @param connection
* @see #pipeline(LettuceResult)
*/
void onCommand(StatefulConnection<?, ?> connection);
/**
* Callback if the pipeline gets closed.
*
* @param connection
* @see #closePipeline()
*/
void onClose(StatefulConnection<?, ?> connection);
}
/**
* Implementation to flush on each command.
*/
private enum FlushEachCommand implements PipeliningFlushPolicy, PipeliningFlushState {
INSTANCE;
@Override
public PipeliningFlushState newPipeline() {
return INSTANCE;
}
@Override
public void onOpen(StatefulConnection<?, ?> connection) {}
@Override
public void onCommand(StatefulConnection<?, ?> connection) {}
@Override
public void onClose(StatefulConnection<?, ?> connection) {}
}
/**
* Implementation to flush on closing the pipeline.
*/
private enum FlushOnClose implements PipeliningFlushPolicy, PipeliningFlushState {
INSTANCE;
@Override
public PipeliningFlushState newPipeline() {
return INSTANCE;
}
@Override
public void onOpen(StatefulConnection<?, ?> connection) {
connection.setAutoFlushCommands(false);
}
@Override
public void onCommand(StatefulConnection<?, ?> connection) {
}
@Override
public void onClose(StatefulConnection<?, ?> connection) {
connection.setAutoFlushCommands(true);
connection.flushCommands();
}
}
/**
* Pipeline state for buffered flushing.
*/
private static class BufferedFlushing implements PipeliningFlushState {
private final AtomicLong commands = new AtomicLong();
private final int flushAfter;
public BufferedFlushing(int flushAfter) {
this.flushAfter = flushAfter;
}
@Override
public void onOpen(StatefulConnection<?, ?> connection) {
connection.setAutoFlushCommands(false);
}
@Override
public void onCommand(StatefulConnection<?, ?> connection) {
if (commands.incrementAndGet() % flushAfter == 0) {
connection.flushCommands();
}
}
@Override
public void onClose(StatefulConnection<?, ?> connection) {
connection.setAutoFlushCommands(true);
connection.flushCommands();
}
}
}

28
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java
View File

@@ -114,6 +114,7 @@ public class LettuceConnectionFactory
private boolean convertPipelineAndTxResults = true;
private RedisStandaloneConfiguration standaloneConfig = new RedisStandaloneConfiguration("localhost", 6379);
private PipeliningFlushPolicy pipeliningFlushPolicy = PipeliningFlushPolicy.flushEachCommand();
private @Nullable RedisConfiguration configuration;
@@ -393,7 +394,10 @@ public class LettuceConnectionFactory
@Nullable StatefulRedisConnection<byte[], byte[]> sharedConnection, LettuceConnectionProvider connectionProvider,
long timeout, int database) {
return new LettuceConnection(sharedConnection, connectionProvider, timeout, database);
LettuceConnection connection = new LettuceConnection(sharedConnection, connectionProvider, timeout, database);
connection.setPipeliningFlushPolicy(this.pipeliningFlushPolicy);
return connection;
}
/**
@@ -414,8 +418,11 @@ public class LettuceConnectionFactory
LettuceConnectionProvider connectionProvider, ClusterTopologyProvider topologyProvider,
ClusterCommandExecutor clusterCommandExecutor, Duration commandTimeout) {
return new LettuceClusterConnection(sharedConnection, connectionProvider, topologyProvider, clusterCommandExecutor,
commandTimeout);
LettuceClusterConnection connection = new LettuceClusterConnection(sharedConnection, connectionProvider,
topologyProvider, clusterCommandExecutor, commandTimeout);
connection.setPipeliningFlushPolicy(this.pipeliningFlushPolicy);
return connection;
}
/*
@@ -552,6 +559,21 @@ public class LettuceConnectionFactory
standaloneConfig.setPort(port);
}
/**
* Configures the flushing policy when using pipelining. Defaults to {@link PipeliningFlushPolicy#flushEachCommand()
* flush on each command}.
*
* @param pipeliningFlushPolicy the flushing policy to control when commands get written to the Redis connection.
* @see LettuceConnection#openPipeline()
* @see StatefulRedisConnection#flushCommands()
*/
public void setPipeliningFlushPolicy(PipeliningFlushPolicy pipeliningFlushPolicy) {
Assert.notNull(pipeliningFlushPolicy, "PipeliningFlushingPolicy must not be null!");
this.pipeliningFlushPolicy = pipeliningFlushPolicy;
}
/**
* Returns the connection timeout (in milliseconds).
*