From b656b09db59b4e0577058cb02a0f291a0bc6abaf Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Mon, 18 Dec 2017 10:13:14 +0100 Subject: [PATCH] DATAREDIS-731 - Share native Lettuce Cluster connection across LettuceClusterConnection. We now share Lettuce Cluster connections if configured on LettuceConnectionFactory (enabled by default) across multiple LettuceClusterConnections. Connection sharing prevents creation and disposal of cluster connections per template command execution which improves overall performance. Original Pull Request: #297 --- .../lettuce/LettuceClusterConnection.java | 23 ++++++++- .../connection/lettuce/LettuceConnection.java | 13 +++++ .../lettuce/LettuceConnectionFactory.java | 13 +++-- .../LettuceClusterConnectionUnitTests.java | 48 +++++++++++++++++++ .../LettuceConnectionFactoryUnitTests.java | 34 +++++++++++++ 5 files changed, 123 insertions(+), 8 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java index c10a33e3f..8c8eff434 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java @@ -152,6 +152,28 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau this.disposeClusterCommandExecutorOnClose = false; } + /** + * Creates new {@link LettuceClusterConnection} given a shared {@link StatefulRedisClusterConnection} + * and{@link LettuceConnectionProvider} running commands across the cluster via given {@link ClusterCommandExecutor}. + * + * @param sharedConnection must not be {@literal null}. + * @param connectionProvider must not be {@literal null}. + * @param executor must not be {@literal null}. + * @param timeout must not be {@literal null}. + * @since 2.1 + */ + public LettuceClusterConnection(StatefulRedisClusterConnection sharedConnection, + LettuceConnectionProvider connectionProvider, ClusterCommandExecutor executor, Duration timeout) { + + super(sharedConnection, connectionProvider, timeout.toMillis(), 0); + + Assert.notNull(executor, "ClusterCommandExecutor must not be null."); + + this.topologyProvider = new LettuceClusterTopologyProvider(getClient()); + this.clusterCommandExecutor = executor; + this.disposeClusterCommandExecutorOnClose = false; + } + /** * @return access to {@link RedisClusterClient} for non-connection access. */ @@ -319,7 +341,6 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau clusterCommandExecutor.executeCommandOnSingleNode( (LettuceClusterCommandCallback) client -> client.clusterAddSlots(slots), node); - } /* diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java index c09ff5c06..751c0fc48 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java @@ -243,6 +243,19 @@ public class LettuceConnection extends AbstractRedisConnection { */ public LettuceConnection(StatefulRedisConnection sharedConnection, LettuceConnectionProvider connectionProvider, long timeout, int defaultDbIndex) { + this((StatefulConnection) sharedConnection, connectionProvider, timeout, defaultDbIndex); + } + + /** + * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. Should not be + * used for transactions or blocking operations. + * @param connectionProvider connection provider to obtain and release native connections. + * @param timeout The connection timeout (in milliseconds) + * @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection. + * @since 2.1 + */ + LettuceConnection(StatefulConnection sharedConnection, LettuceConnectionProvider connectionProvider, + long timeout, int defaultDbIndex) { Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null."); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java index 2eb685bbc..12e806cab 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java @@ -345,7 +345,11 @@ public class LettuceConnectionFactory throw new InvalidDataAccessApiUsageException("Cluster is not configured!"); } - return new LettuceClusterConnection(connectionProvider, clusterCommandExecutor, + return getShareNativeConnection() + ? new LettuceClusterConnection( + (StatefulRedisClusterConnection) getOrCreateSharedConnection().getConnection(), + connectionProvider, clusterCommandExecutor, clientConfiguration.getCommandTimeout()) + : new LettuceClusterConnection(connectionProvider, clusterCommandExecutor, clientConfiguration.getCommandTimeout()); } @@ -921,7 +925,7 @@ public class LettuceConnectionFactory return new StandaloneConnectionProvider((RedisClient) client, codec, readFrom); } - private AbstractRedisClient createClient() { + protected AbstractRedisClient createClient() { if (isRedisSentinelAware()) { @@ -1065,15 +1069,10 @@ public class LettuceConnectionFactory * * @return the connection. */ - @Nullable private StatefulConnection getNativeConnection() { try { - if (isClusterAware() && !shareNativeClusterConnection) { - return null; - } - StatefulConnection connection = connectionProvider.getConnection(StatefulConnection.class); if (connection instanceof StatefulRedisConnection && getDatabase() > 0) { ((StatefulRedisConnection) connection).sync().select(getDatabase()); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java index 0f33ed07a..bd797e25f 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java @@ -24,13 +24,19 @@ import static org.springframework.data.redis.connection.ClusterTestVariables.*; import static org.springframework.data.redis.test.util.MockitoUtils.*; import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulConnection; +import io.lettuce.core.api.StatefulRedisConnection; +import io.lettuce.core.api.sync.RedisCommands; import io.lettuce.core.cluster.RedisClusterClient; +import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; import io.lettuce.core.cluster.api.async.RedisClusterAsyncCommands; +import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands; import io.lettuce.core.cluster.api.sync.RedisClusterCommands; import io.lettuce.core.cluster.models.partitions.Partitions; import io.lettuce.core.cluster.models.partitions.RedisClusterNode.NodeFlag; import java.io.IOException; +import java.time.Duration; import java.util.Arrays; import java.util.Collections; import java.util.List; @@ -62,7 +68,10 @@ public class LettuceClusterConnectionUnitTests { @Mock RedisClusterClient clusterMock; + @Mock ClusterConnectionProvider connectionProviderMock; + @Mock ClusterCommandExecutor executorMock; @Mock ClusterNodeResourceProvider resourceProvider; + @Mock StatefulRedisClusterConnection sharedConnectionMock; @Mock RedisClusterAsyncCommands dedicatedConnectionMock; @Mock RedisClusterCommands clusterConnection1Mock; @Mock RedisClusterCommands clusterConnection2Mock; @@ -346,4 +355,43 @@ public class LettuceClusterConnectionUnitTests { verify(clusterConnection1Mock, never()).configResetstat(); } + @Test // DATAREDIS-731 + public void shouldExecuteOnSharedConnection() { + + RedisAdvancedClusterCommands sync = mock(RedisAdvancedClusterCommands.class); + + when(connectionProviderMock.getClient()).thenReturn(clusterMock); + when(sharedConnectionMock.sync()).thenReturn(sync); + + LettuceClusterConnection connection = new LettuceClusterConnection(sharedConnectionMock, connectionProviderMock, + executorMock, Duration.ZERO); + + connection.keyCommands().del(KEY_1_BYTES); + + verify(sync).del(KEY_1_BYTES); + verify(connectionProviderMock).getClient(); + verifyNoMoreInteractions(connectionProviderMock); + } + + @Test // DATAREDIS-731 + public void shouldExecuteOnDedicatedConnection() { + + RedisCommands sync = mock(RedisCommands.class); + StatefulRedisConnection dedicatedConnection = mock(StatefulRedisConnection.class); + + when(connectionProviderMock.getClient()).thenReturn(clusterMock); + when(connectionProviderMock.getConnection(StatefulConnection.class)).thenReturn(dedicatedConnection); + when(dedicatedConnection.sync()).thenReturn(sync); + + LettuceClusterConnection connection = new LettuceClusterConnection(sharedConnectionMock, connectionProviderMock, + executorMock, Duration.ZERO); + + connection.listCommands().bLPop(1, KEY_1_BYTES); + + verify(sync).blpop(1, KEY_1_BYTES); + verify(connectionProviderMock).getClient(); + verify(connectionProviderMock).getConnection(StatefulConnection.class); + verifyNoMoreInteractions(connectionProviderMock); + verifyZeroInteractions(sharedConnectionMock); + } } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java index 70ef1bf3d..7f218d050 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java @@ -20,6 +20,7 @@ import static org.hamcrest.core.IsEqual.*; import static org.hamcrest.core.IsInstanceOf.*; import static org.hamcrest.core.IsNull.*; import static org.junit.Assert.*; +import static org.mockito.Mockito.*; import static org.springframework.data.redis.connection.ClusterTestVariables.*; import static org.springframework.data.redis.connection.lettuce.LettuceTestClientResources.*; import static org.springframework.test.util.ReflectionTestUtils.*; @@ -29,6 +30,9 @@ import io.lettuce.core.ClientOptions; import io.lettuce.core.RedisClient; import io.lettuce.core.RedisURI; import io.lettuce.core.cluster.RedisClusterClient; +import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; +import io.lettuce.core.codec.ByteArrayCodec; +import io.lettuce.core.codec.RedisCodec; import io.lettuce.core.resource.ClientResources; import java.security.NoSuchAlgorithmException; @@ -38,6 +42,8 @@ import java.util.Collections; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.mockito.ArgumentMatchers; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.data.redis.ConnectionFactoryTracker; import org.springframework.data.redis.connection.RedisClusterConfiguration; import org.springframework.data.redis.connection.RedisClusterConnection; @@ -552,6 +558,7 @@ public class LettuceConnectionFactoryUnitTests { LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfig); connectionFactory.setShutdownTimeout(0); connectionFactory.setTimeout(2000); + connectionFactory.setShareNativeConnection(false); connectionFactory.afterPropertiesSet(); ConnectionFactoryTracker.add(connectionFactory); @@ -566,6 +573,7 @@ public class LettuceConnectionFactoryUnitTests { LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfig, LettuceClientConfiguration .builder().commandTimeout(Duration.ofSeconds(2)).shutdownTimeout(Duration.ZERO).build()); + connectionFactory.setShareNativeConnection(false); connectionFactory.afterPropertiesSet(); ConnectionFactoryTracker.add(connectionFactory); @@ -575,4 +583,30 @@ public class LettuceConnectionFactoryUnitTests { clusterConnection.close(); } + + @Test // DATAREDIS-731 + public void shouldShareNativeConnectionWithCluster() { + + RedisClusterClient clientMock = mock(RedisClusterClient.class); + StatefulRedisClusterConnection connectionMock = mock(StatefulRedisClusterConnection.class); + when(clientMock.connect(ByteArrayCodec.INSTANCE)).thenReturn(connectionMock); + + LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfig, + LettuceClientConfiguration.defaultConfiguration()) { + + @Override + protected AbstractRedisClient createClient() { + return clientMock; + } + }; + + connectionFactory.afterPropertiesSet(); + + new DirectFieldAccessor(connectionFactory).setPropertyValue("client", clientMock); + + connectionFactory.getClusterConnection().close(); + connectionFactory.getClusterConnection().close(); + + verify(clientMock).connect(ArgumentMatchers.any(RedisCodec.class)); + } }