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 fcfa511dc..9a854c4cc 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 @@ -82,6 +82,7 @@ public class LettuceConnection implements RedisConnection { private boolean isClosed = false; private boolean isMulti = false; private boolean isPipelined = false; + private boolean closeNativeConnection = true; private List> ppline; private RedisClient client; private volatile LettuceSubscription subscription; @@ -91,15 +92,31 @@ public class LettuceConnection implements RedisConnection { * * @param connection underlying Lettuce async connection * @param timeout the connection timeout (in milliseconds) + * @param client The Lettuce client */ public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection connection, long timeout, RedisClient client) { + this(connection, timeout, client, true); + } + + /** + * Instantiates a new lettuce connection. + * + * @param connection The underlying Lettuce async connection + * @param timeout The connection timeout (in milliseconds) + * @param client The Lettuce client + * @param closeNativeConnection True if native connection should be called on {@link #close()}. This should + * be set to false if the native connection is shared. + */ + public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection connection, long timeout, + RedisClient client, boolean closeNativeConnection) { Assert.notNull(connection, "a valid connection is required"); this.asyncConn = connection; this.timeout = timeout; this.con = new com.lambdaworks.redis.RedisConnection(asyncConn); this.client = client; + this.closeNativeConnection = closeNativeConnection; } protected DataAccessException convertLettuceAccessException(Exception ex) { @@ -161,6 +178,14 @@ public class LettuceConnection implements RedisConnection { txConn = null; } + if(closeNativeConnection) { + try { + asyncConn.close(); + } catch (RuntimeException ex) { + throw convertLettuceAccessException(ex); + } + } + if (subscription != null) { if(subscription.isAlive()) { subscription.doClose(); 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 779cbe932..4c6bd5f23 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 @@ -37,7 +37,7 @@ import com.lambdaworks.redis.RedisException; *

* This factory creates a new {@link LettuceConnection} on each call to * {@link #getConnection()}. Multiple {@link LettuceConnection}s share a single - * thread-safe native connection. + * thread-safe native connection by default. * *

* The shared native connection is never closed by {@link LettuceConnection}, @@ -57,6 +57,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea private RedisClient client; private long timeout = TimeUnit.MILLISECONDS.convert(60, TimeUnit.SECONDS); private boolean validateConnection = false; + private boolean shareNativeConnection = true; private RedisAsyncConnection connection; private int dbIndex = 0; @@ -79,8 +80,6 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea public void afterPropertiesSet() { client = new RedisClient(hostName, port); client.setDefaultTimeout(timeout, TimeUnit.MILLISECONDS); - // open a single connection to be shared for non-blocking and non-tx ops - initConnection(); } public void destroy() { @@ -88,15 +87,11 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea } public RedisConnection getConnection() { - if (validateConnection) { - try { - new com.lambdaworks.redis.RedisConnection(connection).ping(); - } catch (RedisException e) { - log.warn("Validation of shared connection failed. Creating a new connection."); - initConnection(); - } + RedisAsyncConnection nativeConnection = getNativeConnection(); + if (dbIndex > 0) { + nativeConnection.select(dbIndex); } - return new LettuceConnection(connection, timeout, client); + return new LettuceConnection(nativeConnection, timeout, client, !shareNativeConnection); } public DataAccessException translateExceptionIfPossible(RuntimeException ex) { @@ -175,12 +170,13 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea * validation fails. *

* Lettuce will automatically reconnect until close is called, which should - * never happen through {@link LettuceConnection}, therefore the default is - * false. + * never happen through {@link LettuceConnection} if a shared native + * connection is used, therefore the default is false. *

* Setting this to true will result in a round-trip call to the server on - * each new connection, so this setting should only be used if there is code - * that is actively closing the native Lettuce connection. + * each new connection, so this setting should only be used if connection + * sharing is enabled and there is code that is actively closing the native + * Lettuce connection. * * @param validateConnection * enable connection validation @@ -189,6 +185,28 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea this.validateConnection = validateConnection; } + /** + * Indicates if multiple {@link LettuceConnection}s should share a single + * native connection. + * + * @return native connection shared + */ + public boolean getShareNativeConnection() { + return shareNativeConnection; + } + + /** + * Enables multiple {@link LettuceConnection}s to share a single native + * connection. If set to false, every operation on {@link LettuceConnection} + * will open and close a socket. + * + * @param shareNativeConnection + * enable connection sharing + */ + public void setShareNativeConnection(boolean shareNativeConnection) { + this.shareNativeConnection = shareNativeConnection; + } + /** * Returns the index of the database. * @@ -210,10 +228,22 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea this.dbIndex = index; } - protected void initConnection() { - connection = client.connectAsync(LettuceUtils.CODEC); - if (dbIndex > 0) { - connection.select(dbIndex); + protected RedisAsyncConnection getNativeConnection() { + if(shareNativeConnection) { + if(connection == null) { + connection = client.connectAsync(LettuceUtils.CODEC); + } + if (validateConnection) { + try { + new com.lambdaworks.redis.RedisConnection(connection).ping(); + } catch (RedisException e) { + log.warn("Validation of shared connection failed. Creating a new connection."); + connection = client.connectAsync(LettuceUtils.CODEC); + } + } + return connection; + }else { + return client.connectAsync(LettuceUtils.CODEC); } } } \ No newline at end of file diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java index 3a1f3d2d0..8687550e2 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java @@ -18,6 +18,7 @@ package org.springframework.data.redis.connection.lettuce; import static org.junit.Assert.assertEquals; import static org.junit.Assert.fail; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.assertNotSame; import org.junit.After; import org.junit.Before; @@ -29,6 +30,7 @@ import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.StringRedisConnection; import com.lambdaworks.redis.RedisAsyncConnection; +import com.lambdaworks.redis.RedisException; /** * Integration test of {@link LettuceConnectionFactory} @@ -119,4 +121,24 @@ public class LettuceConnectionFactoryTests { factory2.destroy(); } } + + @SuppressWarnings("unchecked") + @Test + public void testDisableSharedConnection() throws Exception { + factory.setShareNativeConnection(false); + RedisConnection conn2 = factory.getConnection(); + assertNotSame(connection.getNativeConnection(), conn2.getNativeConnection()); + // Give some time for native connection to asynchronously initialize, else close doesn't work + Thread.sleep(100); + conn2.close(); + assertTrue(conn2.isClosed()); + // Give some time for native connection to asynchronously close + Thread.sleep(100); + try { + ((RedisAsyncConnection) conn2.getNativeConnection()).ping(); + fail("The native connection should be closed"); + } catch (RedisException e) { + // expected + } + } }