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

DATAREDIS-545 - Support connection-pooling for Redis Cluster with Lettuce.

Browse Source 
We now support connection pooling for Cluster connections using Lettuce.

Original Pull Request: #297
This commit is contained in:
Mark Paluch
2017-12-18 14:23:38 +01:00
committed by Christoph Strobl
parent a3d4e4f771
commit 155f45c376
5 changed files with 62 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -22,6 +22,7 @@ import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.SlotHash;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.sync.RedisClusterCommands;
import io.lettuce.core.cluster.models.partitions.Partitions;
import lombok.RequiredArgsConstructor;
import java.time.Duration;
@@ -63,6 +64,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
new LettuceExceptionConverter());
private final Log log = LogFactory.getLog(getClass());
private final RedisClusterClient clusterClient;
private ClusterCommandExecutor clusterCommandExecutor;
private ClusterTopologyProvider topologyProvider;
@@ -114,10 +116,14 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
super(null, connectionProvider, RedisURI.DEFAULT_TIMEOUT_DURATION.toMillis(), 0);
topologyProvider = new LettuceClusterTopologyProvider(getClient());
clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider,
Assert.isTrue(connectionProvider instanceof ClusterConnectionProvider,
"LettuceConnectionProvider must be a ClusterConnectionProvider.");
this.clusterClient = ((ClusterConnectionProvider) connectionProvider).getRedisClient();
this.topologyProvider = new LettuceClusterTopologyProvider(this.clusterClient);
this.clusterCommandExecutor = new ClusterCommandExecutor(this.topologyProvider,
new LettuceClusterNodeResourceProvider(getConnectionProvider()), exceptionConverter);
disposeClusterCommandExecutorOnClose = true;
this.disposeClusterCommandExecutorOnClose = true;
}
/**
@@ -147,30 +153,37 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
super(null, connectionProvider, timeout.toMillis(), 0);
Assert.notNull(executor, "ClusterCommandExecutor must not be null.");
Assert.isTrue(connectionProvider instanceof ClusterConnectionProvider,
"LettuceConnectionProvider must be a ClusterConnectionProvider.");
this.topologyProvider = new LettuceClusterTopologyProvider(getClient());
this.clusterClient = ((ClusterConnectionProvider) connectionProvider).getRedisClient();
this.topologyProvider = new LettuceClusterTopologyProvider(this.clusterClient);
this.clusterCommandExecutor = executor;
this.disposeClusterCommandExecutorOnClose = false;
}
/**
* Creates new {@link LettuceClusterConnection} given a shared {@link StatefulRedisClusterConnection}
* and{@link LettuceConnectionProvider} running commands across the cluster via given {@link ClusterCommandExecutor}.
* 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 sharedConnection may be {@literal null} if no shared connection used.
* @param connectionProvider must not be {@literal null}.
* @param clusterClient 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<byte[], byte[]> sharedConnection,
LettuceConnectionProvider connectionProvider, ClusterCommandExecutor executor, Duration timeout) {
LettuceClusterConnection(@Nullable StatefulRedisClusterConnection<byte[], byte[]> sharedConnection,
LettuceConnectionProvider connectionProvider, RedisClusterClient clusterClient, ClusterCommandExecutor executor,
Duration timeout) {
super(sharedConnection, connectionProvider, timeout.toMillis(), 0);
Assert.notNull(executor, "ClusterCommandExecutor must not be null.");
Assert.notNull(clusterClient, "RedisClusterClient must not be null.");
this.topologyProvider = new LettuceClusterTopologyProvider(getClient());
this.clusterClient = clusterClient;
this.topologyProvider = new LettuceClusterTopologyProvider(clusterClient);
this.clusterCommandExecutor = executor;
this.disposeClusterCommandExecutorOnClose = false;
}
@@ -190,6 +203,14 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
connectionProvider.getClass().getName()));
}
/**
* @return access to {@link RedisClusterClient} for non-connection access.
*/
private Partitions getPartitions() {
return clusterClient.getPartitions();
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceConnection#geoCommands()
@@ -308,7 +329,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
@Override
public RedisClusterNode clusterGetNodeForSlot(int slot) {
return LettuceConverters.toRedisClusterNode(getClient().getPartitions().getPartitionBySlot(slot));
return LettuceConverters.toRedisClusterNode(getPartitions().getPartitionBySlot(slot));
}
/*
@@ -515,7 +536,7 @@ 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)
*/
@@ -524,7 +545,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
return doGetClusterKeyCommands().scan(node, options);
}
/*
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisClusterConnection#randomKey(org.springframework.data.redis.connection.RedisClusterNode)
*/
@@ -552,7 +573,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
*/
@Override
public List<RedisClusterNode> clusterGetNodes() {
return LettuceConverters.partitionsToClusterNodes(getClient().getPartitions());
return LettuceConverters.partitionsToClusterNodes(getPartitions());
}
/*

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

@@ -241,7 +241,7 @@ public class LettuceConnection extends AbstractRedisConnection {
* @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection.
* @since 2.0
*/
public LettuceConnection(StatefulRedisConnection<byte[], byte[]> sharedConnection,
public LettuceConnection(@Nullable StatefulRedisConnection<byte[], byte[]> sharedConnection,
LettuceConnectionProvider connectionProvider, long timeout, int defaultDbIndex) {
this((StatefulConnection<byte[], byte[]>) sharedConnection, connectionProvider, timeout, defaultDbIndex);
}
@@ -254,7 +254,8 @@ public class LettuceConnection extends AbstractRedisConnection {
* @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection.
* @since 2.1
*/
LettuceConnection(StatefulConnection<byte[], byte[]> sharedConnection, LettuceConnectionProvider connectionProvider,
LettuceConnection(@Nullable StatefulConnection<byte[], byte[]> sharedConnection,
LettuceConnectionProvider connectionProvider,
long timeout, int defaultDbIndex) {
Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null.");

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

@@ -345,12 +345,14 @@ public class LettuceConnectionFactory
throw new InvalidDataAccessApiUsageException("Cluster is not configured!");
}
RedisClusterClient clusterClient = (RedisClusterClient) client;
return getShareNativeConnection()
? new LettuceClusterConnection(
(StatefulRedisClusterConnection<byte[], byte[]>) getOrCreateSharedConnection().getConnection(),
connectionProvider, clusterCommandExecutor, clientConfiguration.getCommandTimeout())
: new LettuceClusterConnection(connectionProvider, clusterCommandExecutor,
clientConfiguration.getCommandTimeout());
connectionProvider, clusterClient, clusterCommandExecutor, clientConfiguration.getCommandTimeout())
: new LettuceClusterConnection(null, connectionProvider, clusterClient, clusterCommandExecutor,
clientConfiguration.getCommandTimeout());
}
/*

14
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java
View File

@@ -68,7 +68,7 @@ public class LettuceClusterConnectionUnitTests {
@Mock RedisClusterClient clusterMock;
@Mock ClusterConnectionProvider connectionProviderMock;
@Mock LettuceConnectionProvider connectionProviderMock;
@Mock ClusterCommandExecutor executorMock;
@Mock ClusterNodeResourceProvider resourceProvider;
@Mock StatefulRedisClusterConnection<byte[], byte[]> sharedConnectionMock;
@@ -355,41 +355,37 @@ public class LettuceClusterConnectionUnitTests {
verify(clusterConnection1Mock, never()).configResetstat();
}
@Test // DATAREDIS-731
@Test // DATAREDIS-731, DATAREDIS-545
public void shouldExecuteOnSharedConnection() {
RedisAdvancedClusterCommands<byte[], byte[]> sync = mock(RedisAdvancedClusterCommands.class);
when(connectionProviderMock.getClient()).thenReturn(clusterMock);
when(sharedConnectionMock.sync()).thenReturn(sync);
LettuceClusterConnection connection = new LettuceClusterConnection(sharedConnectionMock, connectionProviderMock,
executorMock, Duration.ZERO);
clusterMock, executorMock, Duration.ZERO);
connection.keyCommands().del(KEY_1_BYTES);
verify(sync).del(KEY_1_BYTES);
verify(connectionProviderMock).getClient();
verifyNoMoreInteractions(connectionProviderMock);
}
@Test // DATAREDIS-731
@Test // DATAREDIS-731, DATAREDIS-545
public void shouldExecuteOnDedicatedConnection() {
RedisCommands<byte[], byte[]> sync = mock(RedisCommands.class);
StatefulRedisConnection<byte[], byte[]> 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);
clusterMock, 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);

15
src/test/java/org/springframework/data/redis/core/RedisClusterTemplateTests.java
View File

@@ -33,6 +33,7 @@ import org.springframework.data.redis.StringObjectFactory;
import org.springframework.data.redis.connection.RedisClusterConfiguration;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceTestClientResources;
import org.springframework.data.redis.serializer.GenericToStringSerializer;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
@@ -203,6 +204,12 @@ public class RedisClusterTemplateTests<K, V> extends RedisTemplateTests<K, V> {
lettuceConnectionFactory.afterPropertiesSet();
LettuceConnectionFactory pooledLettuceConnectionFactory = new LettuceConnectionFactory(
new RedisClusterConfiguration(CLUSTER_NODES), LettucePoolingClientConfiguration.builder()
.clientResources(LettuceTestClientResources.getSharedClientResources()).build());
pooledLettuceConnectionFactory.afterPropertiesSet();
RedisTemplate<String, String> lettuceStringTemplate = new RedisTemplate<>();
lettuceStringTemplate.setDefaultSerializer(StringRedisSerializer.UTF_8);
lettuceStringTemplate.setConnectionFactory(lettuceConnectionFactory);
@@ -233,6 +240,11 @@ public class RedisClusterTemplateTests<K, V> extends RedisTemplateTests<K, V> {
lettuceJackson2JsonPersonTemplate.setValueSerializer(jackson2JsonSerializer);
lettuceJackson2JsonPersonTemplate.afterPropertiesSet();
RedisTemplate<String, String> pooledLettuceStringTemplate = new RedisTemplate<>();
pooledLettuceStringTemplate.setDefaultSerializer(StringRedisSerializer.UTF_8);
pooledLettuceStringTemplate.setConnectionFactory(pooledLettuceConnectionFactory);
pooledLettuceStringTemplate.afterPropertiesSet();
return Arrays.asList(new Object[][] { //
// JEDIS
@@ -249,7 +261,8 @@ public class RedisClusterTemplateTests<K, V> extends RedisTemplateTests<K, V> {
{ lettuceRawTemplate, rawFactory, rawFactory }, //
{ lettucePersonTemplate, stringFactory, personFactory }, //
{ lettuceXstreamStringTemplate, stringFactory, stringFactory }, //
{ lettuceJackson2JsonPersonTemplate, stringFactory, personFactory } //
{ lettuceJackson2JsonPersonTemplate, stringFactory, personFactory }, //
{ pooledLettuceStringTemplate, stringFactory, stringFactory } //
});
}