今天看了一下Lettuce的源码，就学习一下这个流程，简单说明一下，这个流程是比较复杂。
对于这部分的学习，我是通过源码的ConnectToRedisCluster.java入手。
直奔主题
public class ConnectToRedisCluster {
public static void main(String[] args) {
// Syntax: redis://[password@]host[:port]
RedisClusterClient redisClient = RedisClusterClient.create("redis://password@localhost:7379");
StatefulRedisClusterConnection<String, String> connection = redisClient.connect();
System.out.println("Connected to Redis");
connection.sync().get("121212");
connection.close();
redisClient.shutdown();
}
}
这里氛围两部分，一部分是连接部分，一部分是下面的使用部分。关于链接部分，在下一篇文章中进行一个源码分析。这里大致介绍一下，就是获取Redis的拓扑图，然后解析节点，并且后台启动一个调度任务，定时的进行刷新分区的状态。
关于connection.synce()这一步的动作，我们先来看看StatefulRedisClusterConnection的构造方法。
public StatefulRedisClusterConnectionImpl(RedisChannelWriter writer, RedisCodec<K, V> codec, Duration timeout) {
super(writer, timeout);
this.codec = codec;
this.async = new RedisAdvancedClusterAsyncCommandsImpl<>(this, codec);
// 进行一个动态代理.
this.sync = (RedisAdvancedClusterCommands) Proxy.newProxyInstance(AbstractRedisClient.class.getClassLoader(),
new Class<?>[] { RedisAdvancedClusterCommands.class }, syncInvocationHandler());
this.reactive = new RedisAdvancedClusterReactiveCommandsImpl<>(this, codec);
}
没错，sync是一个通过JDK动态代理得到的一个结果对象。那么他代理的对象又是谁，又是怎么操作的呢。既然是JDK的Proxy。那么我们就要看看他的InvocationHandler了。点击syncInvocationHandler.
protected InvocationHandler syncInvocationHandler() {
return new ClusterFutureSyncInvocationHandler<>(this, RedisClusterAsyncCommands.class, NodeSelection.class,
NodeSelectionCommands.class, async());
}
既然这样，那就继续看InvocationHandler的invoke方法了。跟下来，我们看到了这样一行代码:
Object result = targetMethod.invoke(asyncApi, args);
那asyncApi又是什么，这个时候，我们在返回回去看InvocationHandler的调用构造的地方syncInvocationHandler().可以看到，传递的参数给asyncApi正是async()方法，那async()方法又是返回一个RedisAdvancedClusterAsyncCommandsImpl。那么我们的get方法的实现，想必就是和RedisAdvancedClusterAsyncCommandsImpl有关。通过RedisAdvancedClusterAsyncCommandsImpl的基类，AbstractRedisAsyncCommands，找到了对应的get方法。
@Override
public RedisFuture<V> get(K key) {
return dispatch(commandBuilder.get(key));
}
我们继续一路往下走，直到到了ClusterDistributionChannelWriter的doWrite方法:
@Override
public <K, V, T> RedisCommand<K, V, T> write(RedisCommand<K, V, T> command) {
LettuceAssert.notNull(command, "Command must not be null");
if (closed) {
throw new RedisException("Connection is closed");
}
return doWrite(command);
}
private <K, V, T> RedisCommand<K, V, T> doWrite(RedisCommand<K, V, T> command) {
if (command instanceof ClusterCommand && !command.isDone()) {
ClusterCommand<K, V, T> clusterCommand = (ClusterCommand<K, V, T>) command;
if (clusterCommand.isMoved() || clusterCommand.isAsk()) {
HostAndPort target;
boolean asking;
if (clusterCommand.isMoved()) {
target = getMoveTarget(clusterCommand.getError());
clusterEventListener.onMovedRedirection();
asking = false;
} else {
target = getAskTarget(clusterCommand.getError());
asking = true;
clusterEventListener.onAskRedirection();
}
command.getOutput().setError((String) null);
CompletableFuture<StatefulRedisConnection<K, V>> connectFuture = asyncClusterConnectionProvider
.getConnectionAsync(Intent.WRITE, target.getHostText(), target.getPort());
if (isSuccessfullyCompleted(connectFuture)) {
writeCommand(command, asking, connectFuture.join(), null);
} else {
connectFuture.whenComplete((connection, throwable) -> writeCommand(command, asking, connection, throwable));
}
return command;
}
}
ClusterCommand<K, V, T> commandToSend = getCommandToSend(command);
CommandArgs<K, V> args = command.getArgs();
// exclude CLIENT commands from cluster routing
if (args != null && !CommandType.CLIENT.equals(commandToSend.getType())) {
ByteBuffer encodedKey = args.getFirstEncodedKey();
if (encodedKey != null) {
int hash = getSlot(encodedKey);
Intent intent = getIntent(command.getType());
CompletableFuture<StatefulRedisConnection<K, V>> connectFuture = ((AsyncClusterConnectionProvider) clusterConnectionProvider)
.getConnectionAsync(intent, hash);
if (isSuccessfullyCompleted(connectFuture)) {
writeCommand(commandToSend, false, connectFuture.join(), null);
} else {
connectFuture.whenComplete((connection, throwable) -> writeCommand(commandToSend, false, connection,
throwable));
}
return commandToSend;
}
}
writeCommand(commandToSend, defaultWriter);
return commandToSend;
}
我们重点关注这里的代码:
int hash = getSlot(encodedKey);
CompletableFuture<StatefulRedisConnection<K, V>> connectFuture = ((AsyncClusterConnectionProvider) clusterConnectionProvider)
.getConnectionAsync(intent, hash);
相比看到这里就大致直到了他是在干什么的的，getSlot就是获取key对应的slot。那么我们知道了Slot，就很快能知道对应的分区Partition,找到了Partition，我们就能获取到了对应的host和port等信息，然后就可以发送请求到对应的Redis实例进行操作啦。
getConnectionAsync的里面的代码供上:
@Override
public CompletableFuture<StatefulRedisConnection<K, V>> getConnectionAsync(Intent intent, int slot) {
if (debugEnabled) {
logger.debug("getConnection(" + intent + ", " + slot + ")");
}
if (intent == Intent.READ && readFrom != null && readFrom != ReadFrom.MASTER) {
return getReadConnection(slot);
}
return getWriteConnection(slot).toCompletableFuture();
}
这里我们且不先关注是READ还是WRITE，两个方法的内部都是有
RedisClusterNode partition = partitions.getPartitionBySlot(slot);
那有了这些信息之后，后面的事情，我们也是能够猜测到，通过partition那host和port，然后调用返回一个Future。而这个Future的最后的数据转换为同步的数据获取，这里就要回到我们的InvocationHandler了。
Object result = targetMethod.invoke(asyncApi, args);
// 对返回的异步结果进行一个读取。
if (result instanceof RedisFuture) {
RedisFuture<?> command = (RedisFuture<?>) result;
if (!method.getName().equals("exec") && !method.getName().equals("multi")) {
if (connection instanceof StatefulRedisConnection && ((StatefulRedisConnection) connection).isMulti()) {
return null;
}
}
return LettuceFutures.awaitOrCancel(command, getTimeoutNs(command), TimeUnit.NANOSECONDS);
}
LettuceFutures.awaitOrCance的里面就是一个对Future的get等待而已啦。
而这些操作，我们看似一个通过的请求，实际是一个内部的通过netty的异步操作，然后再等待转为同步结果的一个过程。这里内部
