Make JedisCluster multihread by improving connection handling
This commit is contained in:
File diff suppressed because it is too large
Load Diff
@@ -23,9 +23,10 @@ public abstract class JedisClusterCommand<T> {
|
|||||||
this.redirections = maxRedirections;
|
this.redirections = maxRedirections;
|
||||||
}
|
}
|
||||||
|
|
||||||
public abstract T execute();
|
public abstract T execute(Jedis connection);
|
||||||
|
|
||||||
public T run(String key) {
|
public T run(String key) {
|
||||||
|
Jedis connection = null;
|
||||||
try {
|
try {
|
||||||
|
|
||||||
if (key == null) {
|
if (key == null) {
|
||||||
@@ -35,16 +36,20 @@ public abstract class JedisClusterCommand<T> {
|
|||||||
throw new JedisClusterMaxRedirectionsException(
|
throw new JedisClusterMaxRedirectionsException(
|
||||||
"Too many Cluster redirections?");
|
"Too many Cluster redirections?");
|
||||||
}
|
}
|
||||||
connectionHandler.getConnectionFromSlot(JedisClusterCRC16
|
connection = connectionHandler.getConnectionFromSlot(JedisClusterCRC16
|
||||||
.getSlot(key));
|
.getSlot(key));
|
||||||
if (asking) {
|
if (asking) {
|
||||||
// TODO: Pipeline asking with the original command to make it
|
// TODO: Pipeline asking with the original command to make it
|
||||||
// faster....
|
// faster....
|
||||||
connectionHandler.getConnection().asking();
|
connection.asking();
|
||||||
}
|
}
|
||||||
return execute();
|
return execute(connection);
|
||||||
} catch (JedisRedirectionException jre) {
|
} catch (JedisRedirectionException jre) {
|
||||||
return handleRedirection(jre, key);
|
return handleRedirection(jre, key);
|
||||||
|
} finally {
|
||||||
|
if (connection != null) {
|
||||||
|
connectionHandler.returnConnection(connection);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -12,6 +12,13 @@ public abstract class JedisClusterConnectionHandler {
|
|||||||
|
|
||||||
abstract Jedis getConnection();
|
abstract Jedis getConnection();
|
||||||
|
|
||||||
|
protected void returnConnection(Jedis connection) {
|
||||||
|
nodes.get(
|
||||||
|
connection.getClient().getHost()
|
||||||
|
+ connection.getClient().getPort()).returnResource(
|
||||||
|
connection);
|
||||||
|
}
|
||||||
|
|
||||||
abstract Jedis getConnectionFromSlot(int slot);
|
abstract Jedis getConnectionFromSlot(int slot);
|
||||||
|
|
||||||
public JedisClusterConnectionHandler(Set<HostAndPort> nodes) {
|
public JedisClusterConnectionHandler(Set<HostAndPort> nodes) {
|
||||||
|
|||||||
@@ -5,25 +5,12 @@ import java.util.Set;
|
|||||||
public class JedisSlotBasedConnectionHandler extends
|
public class JedisSlotBasedConnectionHandler extends
|
||||||
JedisClusterConnectionHandler {
|
JedisClusterConnectionHandler {
|
||||||
|
|
||||||
private Jedis currentConnection;
|
|
||||||
|
|
||||||
public JedisSlotBasedConnectionHandler(Set<HostAndPort> nodes) {
|
public JedisSlotBasedConnectionHandler(Set<HostAndPort> nodes) {
|
||||||
super(nodes);
|
super(nodes);
|
||||||
}
|
}
|
||||||
|
|
||||||
public Jedis getConnection() {
|
public Jedis getConnection() {
|
||||||
return currentConnection != null ? currentConnection
|
return getRandomConnection().getResource();
|
||||||
: getRandomConnection().getResource();
|
|
||||||
}
|
|
||||||
|
|
||||||
private void returnCurrentConnection() {
|
|
||||||
if (currentConnection != null) {
|
|
||||||
nodes.get(
|
|
||||||
currentConnection.getClient().getHost()
|
|
||||||
+ currentConnection.getClient().getPort())
|
|
||||||
.returnResource(currentConnection);
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@@ -34,13 +21,11 @@ public class JedisSlotBasedConnectionHandler extends
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Jedis getConnectionFromSlot(int slot) {
|
public Jedis getConnectionFromSlot(int slot) {
|
||||||
returnCurrentConnection();
|
|
||||||
JedisPool connectionPool = slots.get(slot);
|
JedisPool connectionPool = slots.get(slot);
|
||||||
if (connectionPool == null) {
|
if (connectionPool == null) {
|
||||||
connectionPool = getRandomConnection();
|
connectionPool = getRandomConnection();
|
||||||
}
|
}
|
||||||
currentConnection = connectionPool.getResource();
|
return connectionPool.getResource();
|
||||||
return connectionPool.getResource();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user