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

Lazily instantiate Lettuce dedicated connections

Browse Source
This commit is contained in:
Jennifer Hickey
2013-07-01 17:28:00 -07:00
parent a0026ebc90
commit 687f5f0d2c
3 changed files with 100 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -88,35 +88,46 @@ public class LettuceConnection implements RedisConnection {
/**
* Instantiates a new lettuce connection.
*
* @param dedicatedConnection
* A dedicated native connection. Can be used for any operation.
* @param timeout
* The connection timeout (in milliseconds)
* @param client
* The {@link RedisClient} to use when making pub/sub connections
* The {@link RedisClient} to use when instantiating a native
* connection
*/
public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection<byte[], byte[]> dedicatedConnection, long timeout,
RedisClient client) {
this(null, dedicatedConnection, timeout, client, null);
public LettuceConnection(long timeout, RedisClient client) {
this(null, timeout, client, null);
}
/**
* Instantiates a new lettuce connection.
*
* @param dedicatedConnection
* A dedicated native connection. Can be used for any operation.
* @param timeout
* The connection timeout (in milliseconds) * @param client The
* {@link RedisClient} to use when instantiating a pub/sub
* connection
* @param pool
* The connection pool to use for all other native connections
*/
public LettuceConnection(long timeout, RedisClient client, LettucePool pool) {
this(null, timeout, client, pool);
}
/**
* Instantiates a new lettuce connection.
*
* @param sharedConnection
* A native connection that is shared with other
* {@link LettuceConnection}s. Will not be used for transactions
* or blocking operations
* @param timeout
* The connection timeout (in milliseconds)
* @param client
* The {@link RedisClient} to use when making pub/sub connections
* @param pool
* An optional connection pool, from which the dedicated
* connection came. If pool is set, the dedicated connection will
* be returned to the pool on close
* The {@link RedisClient} to use when making pub/sub, blocking,
* and tx connections
*/
public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection<byte[], byte[]> dedicatedConnection, long timeout,
RedisClient client, LettucePool pool) {
this(null, dedicatedConnection, timeout, client, pool);
public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection<byte[], byte[]> sharedConnection,
long timeout, RedisClient client) {
this(sharedConnection, timeout, client, null);
}
/**
@@ -126,47 +137,19 @@ public class LettuceConnection implements RedisConnection {
* A native connection that is shared with other
* {@link LettuceConnection}s. Should not be used for
* transactions or blocking operations
* @param dedicatedConnection
* A dedicated native connection. Can be used for any operation.
* @param timeout
* The connection timeout (in milliseconds)
* @param client
* The {@link RedisClient} to use when making pub/sub connections
*/
public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection<byte[], byte[]> sharedConnection,
com.lambdaworks.redis.RedisAsyncConnection<byte[], byte[]> dedicatedConnection, long timeout,
RedisClient client) {
this(sharedConnection, dedicatedConnection, timeout, client, null);
}
/**
* Instantiates a new lettuce connection.
*
* @param sharedConnection
* A native connection that is shared with other
* {@link LettuceConnection}s. Should not be used for
* transactions or blocking operations
* @param dedicatedConnection
* A dedicated native connection. Can be used for any operation.
* @param timeout
* The connection timeout (in milliseconds)
* @param client
* The {@link RedisClient} to use when making pub/sub connections
* The {@link RedisClient} to use when making pub/sub connections
* @param pool
* An optional connection pool, from which the dedicated
* connection came. If pool is set, the dedicated connection will
* be returned to the pool on close
* The connection pool to use for blocking and tx operations
*/
public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection<byte[], byte[]> sharedConnection,
com.lambdaworks.redis.RedisAsyncConnection<byte[], byte[]> dedicatedConnection, long timeout,
RedisClient client, LettucePool pool) {
Assert.notNull(dedicatedConnection, "a valid dedicated connection is required");
long timeout, RedisClient client, LettucePool pool) {
this.asyncSharedConn = sharedConnection;
this.asyncDedicatedConn = dedicatedConnection;
this.timeout = timeout;
this.sharedConn = sharedConnection != null ?
new com.lambdaworks.redis.RedisConnection<byte[], byte[]>(asyncSharedConn) : null;
this.dedicatedConn = new com.lambdaworks.redis.RedisConnection<byte[], byte[]>(dedicatedConnection);
this.sharedConn = sharedConnection != null ? new com.lambdaworks.redis.RedisConnection<byte[], byte[]>(
asyncSharedConn) : null;
this.client = client;
this.pool = pool;
}
@@ -223,19 +206,20 @@ public class LettuceConnection implements RedisConnection {
public void close() throws DataAccessException {
isClosed = true;
if(pool != null) {
if (!broken) {
pool.returnResource(asyncDedicatedConn);
}else {
pool.returnBrokenResource(asyncDedicatedConn);
if(asyncDedicatedConn != null) {
if(pool != null) {
if (!broken) {
pool.returnResource(asyncDedicatedConn);
}else {
pool.returnBrokenResource(asyncDedicatedConn);
}
} else {
try {
asyncDedicatedConn.close();
} catch (RuntimeException ex) {
throw convertLettuceAccessException(ex);
}
}
return;
}
try {
asyncDedicatedConn.close();
} catch (RuntimeException ex) {
throw convertLettuceAccessException(ex);
}
if (subscription != null) {
@@ -549,30 +533,28 @@ public class LettuceConnection implements RedisConnection {
}
}
public void discard() {
isMulti = false;
try {
if (isPipelined()) {
pipeline(getAsyncTxConnection().discard());
pipeline(getAsyncDedicatedConnection().discard());
return;
}
getTxConnection().discard();
getDedicatedConnection().discard();
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
}
public List<Object> exec() {
isMulti = false;
try {
if (isPipelined()) {
getAsyncTxConnection().exec();
return null;
getAsyncDedicatedConnection().exec();
return null;
}
List<Object> results = getTxConnection().exec();
if(results.isEmpty()) {
List<Object> results = getDedicatedConnection().exec();
if (results.isEmpty()) {
return null;
}
return results;
@@ -701,16 +683,15 @@ public class LettuceConnection implements RedisConnection {
isMulti = true;
try {
if (isPipelined()) {
pipeline(getAsyncTxConnection().multi());
pipeline(getAsyncDedicatedConnection().multi());
return;
}
getTxConnection().multi();
getDedicatedConnection().multi();
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
}
public Boolean persist(byte[] key) {
try {
if (isPipelined()) {
@@ -816,28 +797,25 @@ public class LettuceConnection implements RedisConnection {
}
}
public void unwatch() {
try {
if (isPipelined()) {
pipeline(getAsyncTxConnection().unwatch());
pipeline(getAsyncDedicatedConnection().unwatch());
return;
}
getTxConnection().unwatch();
getDedicatedConnection().unwatch();
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
}
public void watch(byte[]... keys) {
try {
if (isPipelined()) {
pipeline(getAsyncTxConnection().watch(keys));
pipeline(getAsyncDedicatedConnection().watch(keys));
return;
}
else {
getTxConnection().watch(keys);
} else {
getDedicatedConnection().watch(keys);
}
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
@@ -1159,10 +1137,10 @@ public class LettuceConnection implements RedisConnection {
public List<byte[]> bLPop(int timeout, byte[]... keys) {
try {
if (isPipelined()) {
pipeline(getAsyncBlockingConnection().blpop(timeout, keys));
pipeline(getAsyncDedicatedConnection().blpop(timeout, keys));
return null;
}
return LettuceUtils.toList(getBlockingConnection().blpop(timeout, keys));
return LettuceUtils.toList(getDedicatedConnection().blpop(timeout, keys));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
@@ -1171,16 +1149,15 @@ public class LettuceConnection implements RedisConnection {
public List<byte[]> bRPop(int timeout, byte[]... keys) {
try {
if (isPipelined()) {
pipeline(getAsyncBlockingConnection().brpop(timeout, keys));
pipeline(getAsyncDedicatedConnection().brpop(timeout, keys));
return null;
}
return LettuceUtils.toList(getBlockingConnection().brpop(timeout, keys));
return LettuceUtils.toList(getDedicatedConnection().brpop(timeout, keys));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
}
public byte[] lIndex(byte[] key, long index) {
try {
if (isPipelined()) {
@@ -1307,10 +1284,10 @@ public class LettuceConnection implements RedisConnection {
public byte[] bRPopLPush(int timeout, byte[] srcKey, byte[] dstKey) {
try {
if (isPipelined()) {
pipeline(getAsyncBlockingConnection().brpoplpush(timeout, srcKey, dstKey));
pipeline(getAsyncDedicatedConnection().brpoplpush(timeout, srcKey, dstKey));
return null;
}
return getBlockingConnection().brpoplpush(timeout, srcKey, dstKey);
return getDedicatedConnection().brpoplpush(timeout, srcKey, dstKey);
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
@@ -2220,42 +2197,43 @@ public class LettuceConnection implements RedisConnection {
}
private RedisAsyncConnection<byte[], byte[]> getAsyncConnection() {
if(isQueueing()) {
return getAsyncTxConnection();
if (isQueueing()) {
return getAsyncDedicatedConnection();
}
if(asyncSharedConn != null) {
if (asyncSharedConn != null) {
return asyncSharedConn;
}
return asyncDedicatedConn;
return getAsyncDedicatedConnection();
}
private com.lambdaworks.redis.RedisConnection<byte[], byte[]> getConnection() {
if(isQueueing()) {
return getTxConnection();
if (isQueueing()) {
return getDedicatedConnection();
}
if(sharedConn != null) {
if (sharedConn != null) {
return sharedConn;
}
return dedicatedConn;
return getDedicatedConnection();
}
private RedisAsyncConnection<byte[], byte[]> getAsyncBlockingConnection() {
private RedisAsyncConnection<byte[], byte[]> getAsyncDedicatedConnection() {
if(asyncDedicatedConn == null) {
if(this.pool != null) {
this.asyncDedicatedConn = pool.getResource();
} else {
this.asyncDedicatedConn = client.connectAsync(LettuceUtils.CODEC);
}
}
return asyncDedicatedConn;
}
private com.lambdaworks.redis.RedisConnection<byte[], byte[]> getBlockingConnection() {
private com.lambdaworks.redis.RedisConnection<byte[], byte[]> getDedicatedConnection() {
if(dedicatedConn == null) {
this.dedicatedConn = new com.lambdaworks.redis.RedisConnection<byte[], byte[]>(getAsyncDedicatedConnection());
}
return dedicatedConn;
}
private RedisAsyncConnection<byte[], byte[]> getAsyncTxConnection() {
return asyncDedicatedConn;
}
private com.lambdaworks.redis.RedisConnection<byte[], byte[]> getTxConnection() {
return dedicatedConn;
}
private Future<Long> asyncBitOp(BitOperation op, byte[] destination, byte[]... keys) {
switch (op) {
case AND:

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

@@ -106,7 +106,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
}
public RedisConnection getConnection() {
return new LettuceConnection(getSharedConnection(), createLettuceConnector(false), timeout, client, pool);
return new LettuceConnection(getSharedConnection(), timeout, client, pool);
}
public void initConnection() {
@@ -114,7 +114,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
if (this.connection != null) {
resetConnection();
}
this.connection = createLettuceConnector(true);
this.connection = createLettuceConnector();
}
}
@@ -313,16 +313,13 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
}
}
protected RedisAsyncConnection<byte[], byte[]> createLettuceConnector(boolean shared) {
if(pool != null && !shared) {
return pool.getResource();
}
protected RedisAsyncConnection<byte[], byte[]> createLettuceConnector() {
try {
RedisAsyncConnection<byte[], byte[]> dedicatedConnection = client.connectAsync(LettuceUtils.CODEC);
RedisAsyncConnection<byte[], byte[]> connection = client.connectAsync(LettuceUtils.CODEC);
if(dbIndex > 0) {
dedicatedConnection.select(dbIndex);
connection.select(dbIndex);
}
return dedicatedConnection;
return connection;
} catch (RedisException e) {
throw new RedisConnectionFailureException("Unable to connect to Redis on " +
getHostName() + ":" + getPort(), e);

37
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java
View File

@@ -16,13 +16,12 @@
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 static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import org.apache.commons.pool.impl.GenericObjectPool.Config;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
@@ -31,7 +30,6 @@ import org.springframework.data.redis.RedisConnectionFailureException;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.SettingsUtils;
import org.springframework.data.redis.connection.DefaultStringRedisConnection;
import org.springframework.data.redis.connection.PoolException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.StringRedisConnection;
@@ -182,14 +180,6 @@ public class LettuceConnectionFactoryTests {
factory.afterPropertiesSet();
}
@Test(expected=RedisConnectionFailureException.class)
public void testGetConnectionNotSharedException() {
factory.setShareNativeConnection(false);
factory.setHostName("fakeHost");
factory.afterPropertiesSet();
factory.getConnection();
}
@Test(expected=RedisConnectionFailureException.class)
public void testGetConnectionSharedException() {
factory.setShareNativeConnection(false);
@@ -217,29 +207,22 @@ public class LettuceConnectionFactoryTests {
}
@Test
public void testCreateLettuceConnectorWithPool() {
Config poolConfig = new Config();
poolConfig.maxActive = 1;
poolConfig.maxWait = 1;
DefaultLettucePool pool = new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort(), poolConfig);
public void testCreateFactoryWithPool() {
DefaultLettucePool pool = new DefaultLettucePool(SettingsUtils.getHost(),
SettingsUtils.getPort());
pool.afterPropertiesSet();
LettuceConnectionFactory factory2 = new LettuceConnectionFactory(pool);
factory2.afterPropertiesSet();
factory2.createLettuceConnector(false);
try {
// We know we are using the pool if we try to get another connection
// when maxActive is set to 1
factory2.createLettuceConnector(false);
fail("Expected Exception retrieving connection from pool");
} catch (PoolException e) {
}
factory2.getConnection();
}
@Ignore("Uncomment this test to manually check connection reuse in a pool scenario")
@Test
public void testLotsOfConnections() throws InterruptedException {
// Running a netstat here should show only the 8 conns from the pool (plus 2 from setUp and 1 from factory2 afterPropertiesSet for shared conn)
final LettuceConnectionFactory factory2 = new LettuceConnectionFactory(new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort()));
DefaultLettucePool pool = new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort());
pool.afterPropertiesSet();
final LettuceConnectionFactory factory2 = new LettuceConnectionFactory(pool);
factory2.afterPropertiesSet();
for(int i=1;i< 1000;i++) {
Thread th = new Thread(new Runnable() {