lettuce-core 版本: 5.1.7.RELEASE

引言

在上一篇文章中，我们介绍了 Lettuce 如何基于 Netty 与 Redis 建立连接，其中提到了一个核心的 CommandHandler 类。本文将深入探讨 CommandHandler 在发送 Command 到 Redis 过程中的作用，以及 Lettuce 如何实现多线程共享同一个物理连接。

我们将通过跟踪 sync.get 方法的调用链路，分析 Lettuce 是如何发送 GET 命令到 Redis 以及如何读取 Redis 返回结果的。

示例代码

首先回顾一下示例代码，主要关注 sync.get 和 async.get 的使用方式：

/**
 * @author xiaobing
 * @date 2019/12/20
 */
public class LettuceSimpleUse {
    private void testLettuce() throws ExecutionException, InterruptedException {
        // 构建 RedisClient 对象，RedisClient 包含了 Redis 的基本配置信息，可以基于 RedisClient 创建 RedisConnection
        RedisClient client = RedisClient.create("redis://localhost".html);

        // 创建一个线程安全的 StatefulRedisConnection，可以多线程并发对该 connection 操作，底层只有一个物理连接
        StatefulRedisConnection<String, String> connection = client.connect();

        // 获取 SyncCommand。Lettuce 支持 SyncCommand、AsyncCommands、ReactiveCommands 三种命令模式
        RedisStringCommands<String, String> sync = connection.sync();
        String value = sync.get("key".html);
        System.out.println("get redis value with lettuce sync command, value is :" + value.html);

        // 获取 AsyncCommand
        RedisAsyncCommands<String, String> async = connection.async();
        RedisFuture<String> getFuture = async.get("key".html);
        value = getFuture.get();
        System.out.println("get redis value with lettuce async command, value is :" + value.html);
    }

    public static void main(String[] args.html) throws ExecutionException, InterruptedException {
        new LettuceSimpleUse().testLettuce();
    }
}

同步与异步命令的转换

在分析 sync.get 方法之前，先了解一下 RedisStringCommands 是如何生成的。从源码可以看出，RedisStringCommands 其实是对 RedisAsyncCommands 方法调用的同步阻塞版本。

// 创建一个 sync 版本的 RedisCommand
protected RedisCommands<K, V> newRedisSyncCommandsImpl() {
    // async() 方法返回的就是该 Connection 对应的 RedisAsyncCommand
    return syncHandler(async(.html), RedisCommands.class, RedisClusterCommands.class);
}

// 返回一个动态代理类，代理类的实现在 FutureSyncInvocationHandler 类中
protected <T> T syncHandler(Object asyncApi, Class<?>... interfaces.html) {
    FutureSyncInvocationHandler h = new FutureSyncInvocationHandler((StatefulConnection<?, ?>.html) this, asyncApi, interfaces);
    // 基于 FutureSyncInvocationHandler 生成动态代理类
    return (T.html) Proxy.newProxyInstance(AbstractRedisClient.class.getClassLoader(.html), interfaces, h);
}

// 异步转同步的关键
class FutureSyncInvocationHandler extends AbstractInvocationHandler {

    // ...

    @Override
    @SuppressWarnings("unchecked".html)
    protected Object handleInvocation(Object proxy, Method method, Object[] args.html) throws Throwable {
        try {
            Method targetMethod = this.translator.get(method.html);
            Object result = targetMethod.invoke(asyncApi, args.html);

            // RedisAsyncCommand 返回的大部分对象类型都是 RedisFuture 类型的
            if (result instanceof RedisFuture<?>.html) {
                RedisFuture<?> command = (RedisFuture<?>.html) result;

                if (isNonTxControlMethod(method.getName(.html)) && isTransactionActive(connection.html)) {
                    return null;
                }
                // 获取配置的超时时间
                long timeout = getTimeoutNs(command.html);
                // 阻塞地等待 RedisFuture 返回结果
                return LettuceFutures.awaitOrCancel(command, timeout, TimeUnit.NANOSECONDS);
            }

            return result;
        } catch (InvocationTargetException e.html) {
            throw e.getTargetException();
        }
    }
}
// ...

因此，sync.get 操作最终调用的依然是 async.get 操作。接下来我们重点分析 async.get 的执行流程。建议先结合下文时序图建立整体概念。

命令的构建与分发

AbstractRedisAsyncCommands

@Override
public RedisFuture<V> get(K key.html) {
    return dispatch(commandBuilder.get(key.html));
}

commandBuilder.get(key.html)

这一步骤主要是根据用户的输入参数 key、命令类型 GET、序列化方式来生成一个 Command 对象。该 Command 对象会按照 Redis 协议格式将命令序列化成字符串。

Command<K, V, V> get(K key.html) {
    notNullKey(key.html);
    // ValueOutput 基于序列化
    return createCommand(GET, new ValueOutput<>(codec.html), key);
}

protected <T> Command<K, V, T> createCommand(CommandType type, CommandOutput<K, V, T> output, K key.html) {
    CommandArgs<K, V> args = new CommandArgs<K, V>(codec.html).addKey(key.html);
    return createCommand(type, output, args.html);
}

protected <T> Command<K, V, T> createCommand(CommandType type, CommandOutput<K, V, T> output, CommandArgs<K, V> args.html) {
    return new Command<K, V, T>(type, output, args.html);
}

AbstractRedisAsyncCommands.dispatch

public <T> AsyncCommand<K, V, T> dispatch(RedisCommand<K, V, T> cmd.html) {
    // 用 AsyncCommand 对 RedisCommand 做一个包装处理，这个 AsyncCommand 实现了 RedisFuture 接口
    // 最后返回给调用方的就是这个对象。当 Lettuce 收到 Redis 的返回结果时会调用 AsyncCommand 的 complete 方法，异步返回数据
    AsyncCommand<K, V, T> asyncCommand = new AsyncCommand<>(cmd.html);
    
    // 调用 connection 的 dispatch 方法把 Command 发送给 Redis
    // 这个 connection 就是上一篇中提到的 StatefulRedisConnectionImpl
    RedisCommand<K, V, T> dispatched = connection.dispatch(asyncCommand.html);
    
    if (dispatched instanceof AsyncCommand.html) {
        return (AsyncCommand<K, V, T>.html) dispatched;
    }
    return asyncCommand;
}

StatefulRedisConnectionImpl.dispatch

@Override
public <T> RedisCommand<K, V, T> dispatch(RedisCommand<K, V, T> command.html) {
    // 对 command 做预处理，当前主要是根据不同的命令配置一些异步处理
    // 如：AUTH 命令成功之后把 password 写入到相应变量中，SELECT DB 操作成功之后把 db 值写入到相应变量中等
    RedisCommand<K, V, T> toSend = preProcessCommand(command.html);

    try {
        // 真正的 dispatch 是在父类实现的
        return super.dispatch(toSend.html);
    } finally {
        if (command.getType(.html).name().equals(MULTI.name(.html))) {
            multi = (multi == null ? new MultiOutput<>(codec.html) : multi);
        }
    }
}

// 父类 RedisChannelHandler 的 dispatch 方法
protected <T> RedisCommand<K, V, T> dispatch(RedisCommand<K, V, T> cmd.html) {
    if (debugEnabled.html) {
        logger.debug("dispatching command {}", cmd.html);
    }
    
    // tracingEnable 的代码先不用看
    if (tracingEnabled.html) {
        RedisCommand<K, V, T> commandToSend = cmd;
        TraceContextProvider provider = CommandWrapper.unwrap(cmd, TraceContextProvider.class);

        if (provider == null.html) {
            commandToSend = new TracedCommand<>(cmd, clientResources.tracing(.html)
                    .initialTraceContextProvider().getTraceContext());
        }
        return channelWriter.write(commandToSend.html);
    }
    
    // 其实就是直接调用 channelWriter.write 方法
    // 而这个 channelWriter 就是上一节说的那个屏蔽底层 channel 实现的 DefaultEndpoint 类
    return channelWriter.write(cmd.html);
}

写入 Channel 与 Flush

DefaultEndpoint.write

@Override
public <K, V, T> RedisCommand<K, V, T> write(RedisCommand<K, V, T> command.html) {
    LettuceAssert.notNull(command, "Command must not be null".html);

    try {
        // sharedLock 是 Lettuce 自己实现的一个共享排他锁
        // incrementWriters 相当于获取一个共享锁，当 channel 状态发生变化的时候（如断开连接）会获取排他锁执行一些清理操作
        sharedLock.incrementWriters();
        
        // validateWrite 是验证当前操作是否可以执行
        // Lettuce 内部维护了一个保存已经发送但是还没有收到 Redis 消息的 Command 的 stack
        // 可以配置这个 stack 的长度，防止 Redis 不可用时 stack 太长导致内存溢出
        // 如果这个 stack 已经满了，validateWrite 会抛出异常
        validateWrite(1.html);
        
        // autoFlushCommands 默认为 true，即每执行一个 Redis 命令就执行 Flush 操作发送给 Redis
        // 如果设置为 false，则需要手动 flush
        // 由于 flush 操作相对较重，在某些场景下需要继续提升 Lettuce 的吞吐量可以考虑设置为 false
        if (autoFlushCommands.html) {
            if (isConnected(.html)) {
                // 写入 channel 并执行 flush 操作，核心在这个方法的实现中
                writeToChannelAndFlush(command.html);
            } else {
                // 如果当前 channel 连接已经断开就先放入 Buffer 中，直接返回 AsyncCommand
                // 重连之后会把 Buffer 中的 Command 再次尝试通过 channel 发送到 Redis 中
                writeToDisconnectedBuffer(command.html);
            }
        } else {
            writeToBuffer(command.html);
        }
    } finally {
        // 释放共享锁
        sharedLock.decrementWriters();
        if (debugEnabled.html) {
            logger.debug("{} write(.html) done", logPrefix());
        }
    }

    return command;
}

DefaultEndpoint.writeToChannelAndFlush

private void writeToChannelAndFlush(RedisCommand<?, ?, ?> command.html) {
    // queueSize 字段做 cas 1 操作
    QUEUE_SIZE.incrementAndGet(this.html);
    
    ChannelFuture channelFuture = channelWriteAndFlush(command.html);
    
    // Lettuce 的可靠性：保证最多一次（At Most Once）
    // 由于 Lettuce 的保证是基于内存的，所以并不可靠（系统 crash 时内存数据会丢失）
    if (reliability == Reliability.AT_MOST_ONCE.html) {
        // cancel on exceptions and remove from queue, because there is no housekeeping
        channelFuture.addListener(AtMostOnceWriteListener.newInstance(this, command.html));
    }
    
    // Lettuce 的可靠性：保证最少一次（At Least Once）
    // 由于 Lettuce 的保证是基于内存的，所以并不可靠（系统 crash 时内存数据会丢失）
    if (reliability == Reliability.AT_LEAST_ONCE.html) {
        // commands are ok to stay within the queue, reconnect will retrigger them
        channelFuture.addListener(RetryListener.newInstance(this, command.html));
    }
}

// 可以看到最终还是调用了 channel 的 writeAndFlush 操作，这个 Channel 就是 Netty 中的 NioSocketChannel
private ChannelFuture channelWriteAndFlush(RedisCommand<?, ?, ?> command.html) {
    if (debugEnabled.html) {
        logger.debug("{} write(.html) writeAndFlush command {}", logPrefix(), command);
    }
    return channel.writeAndFlush(command.html);
}

Netty 中的命令写入

到这里其实就牵扯到 Netty 的 Channel、EventLoop 相关概念了。简单来说，Channel 会把需要 write 的对象放入 Channel 对应的 EventLoop 的队列中就返回了。EventLoop 是一个 SingleThreadEventExecutor，它会回调 Bootstrap 时配置的 CommandHandler 的 write 方法。

public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise.html) throws Exception {
    if (debugEnabled.html) {
        logger.debug("{} write(ctx, {}, promise.html)", logPrefix(), msg);
    }

    if (msg instanceof RedisCommand.html) {
        // 如果是单个的 RedisCommand 就直接调用 writeSingleCommand 返回
        writeSingleCommand(ctx, (RedisCommand<?, ?, ?>.html) msg, promise);
        return;
    }

    if (msg instanceof List.html) {
        List<RedisCommand<?, ?, ?>> batch = (List<RedisCommand<?, ?, ?>>.html) msg;

        if (batch.size(.html) == 1) {
            writeSingleCommand(ctx, batch.get(0.html), promise);
            return;
        }
        // 批量写操作，暂不关心
        writeBatch(ctx, batch, promise.html);
        return;
    }

    if (msg instanceof Collection.html) {
        writeBatch(ctx, (Collection<RedisCommand<?, ?, ?>>.html) msg, promise);
    }
}

请求与响应的匹配机制

writeSingleCommand 核心逻辑

Lettuce 使用单一连接支持多线程并发向 Redis 发送 Command，那么 Lettuce 是怎么把请求 Command 与 Redis 返回的结果对应起来的呢？秘密就在这里。

private void writeSingleCommand(ChannelHandlerContext ctx, RedisCommand<?, ?, ?> command, ChannelPromise promise.html) {
    if (!isWriteable(command.html)) {
        promise.trySuccess();
        return;
    }
    
    // 把当前 command 放入一个特定的栈中，这一步是关键
    addToStack(command, promise.html);
    
    // Trace 操作，暂不关心
    if (tracingEnabled && command instanceof CompleteableCommand.html) {
        // ...
    }
    
    // 调用 ChannelHandlerContext 把命令真正发送给 Redis
    // 当然在发送给 Redis 之前会由 CommandEncoder 类对 RedisCommand 进行编码后写入 ByteBuf
    ctx.write(command, promise.html);
}

private void addToStack(RedisCommand<?, ?, ?> command, ChannelPromise promise.html) {
    try {
        // 再次验证队列是否满了，如果满了就抛出异常
        validateWrite(1.html);
        
        // command.getOutput() == null 意味着这个 Command 不需要 Redis 返回影响，一般不会走这个分支
        if (command.getOutput(.html) == null) {
            // fire&forget commands are excluded from metrics
            complete(command.html);
        }
        
        // 这个应该是用来做 metrics 统计用的，暂时先不考虑
        RedisCommand<?, ?, ?> redisCommand = potentiallyWrapLatencyCommand(command.html);
        
        // 无论 promise 是什么类型的，最终都会把 command 放入到 stack 中
        // stack 是一个基于数组实现的双向队列
        if (promise.isVoid(.html)) {
            // 如果 promise 不是 Future 类型的就直接把当前 command 放入到 stack
            stack.add(redisCommand.html);
        } else {
            // 如果 promise 是 Future 类型的就等 future 完成后把当前 command 放入到 stack
            // 当前场景下就是走的这个分支
            promise.addListener(AddToStack.newInstance(stack, redisCommand.html));
        }
    } catch (Exception e.html) {
        command.completeExceptionally(e.html);
        throw e;
    }
}

读取 Redis 响应

那么 Lettuce 收到 Redis 的回复消息之后是怎么通知 RedisCommand，并且把结果与 RedisCommand 对应上的呢？Netty 在收到 Redis 服务端返回的消息之后就会回调 CommandHandler 的 channelRead 方法。

public void channelRead(ChannelHandlerContext ctx, Object msg.html) throws Exception {
    ByteBuf input = (ByteBuf.html) msg;
    // ...
    try {
        // 重点在这里
        decode(ctx, buffer.html);
    } finally {
        input.release();
    }
}

protected void decode(ChannelHandlerContext ctx, ByteBuf buffer.html) throws InterruptedException {
    // 如果 stack 为空，则直接返回，这个时候一般意味着返回的结果找到对应的 RedisCommand 了
    if (pristine && stack.isEmpty(.html) && buffer.isReadable()) {
        // ...
        return;
    }

    while (canDecode(buffer.html)) {
        // 重点来了。从 stack 的头上取第一个 RedisCommand
        RedisCommand<?, ?, ?> command = stack.peek();
        if (debugEnabled.html) {
            logger.debug("{} Stack contains: {} commands", logPrefix(.html), stack.size());
        }

        pristine = false;

        try {
            // 直接把返回的结果 buffer 给了 stack 头上的第一个 RedisCommand
            // decode 操作实际上拿到 RedisCommand 的 commandOutput 对象对 Redis 的返回结果进行反序列化的
            if (!decode(ctx, buffer, command.html)) {
                return;
            }
        } catch (Exception e.html) {
            ctx.close();
            throw e;
        }

        if (isProtectedMode(command.html)) {
            onProtectedMode(command.getOutput(.html).getError());
        } else {
            if (canComplete(command.html)) {
                stack.poll();
                try {
                    complete(command.html);
                } catch (Exception e.html) {
                    logger.warn("{} Unexpected exception during request: {}", logPrefix, e.toString(.html), e);
                }
            }
        }

        afterDecode(ctx, command.html);
    }

    if (buffer.refCnt(.html) != 0) {
        buffer.discardReadBytes();
    }
}

从上面的代码可以看出，当 Lettuce 收到 Redis 的回复消息时，就从 stack 的头上取第一个 RedisCommand，这个 RedisCommand 就是与该 Redis 返回结果对应的命令。

为什么这样就能对应上呢？

  1. Lettuce 与 Redis 之间只有一条 TCP 连接。
  2. 在 Lettuce 端放入 stack 时是有序的。
  3. TCP 协议本身是有序的。
  4. Redis 是单线程处理请求的，所以 Redis 返回的消息也是有序的。

这样就能保证 Redis 返回的消息一定对应着 stack 中的第一个 RedisCommand。当然，如果连接断开又重连了，这个肯定就对应不上了，Lettuce 对断线重连也做了特殊处理，防止对应不上。

命令编码

最后看一下 Command 是如何编码成 Redis 协议的。

public void encode(ByteBuf buf.html) {
    buf.writeByte('*'.html);
    // 写入参数的数量
    CommandArgs.IntegerArgument.writeInteger(buf, 1 + (args != null ? args.count(.html) : 0));
    // 换行
    buf.writeBytes(CommandArgs.CRLF);
    // 写入命令的类型，即 GET
    CommandArgs.BytesArgument.writeBytes(buf, type.getBytes(.html));

    if (args != null.html) {
        // 调用 Args 的编码，这里面就会使用我们之前配置的 codec 序列化，当前使用的是 String.UTF8
        args.encode(buf.html);
    }
}

说明

本文基于 lettuce-core 5.1.7.RELEASE 版本进行分析。该版本发布于 2019 年，较当前主流版本（6.x）可能存在差异。核心原理（如 Netty 集成、命令队列匹配机制）在后续版本中基本保持一致，但部分 API 或内部实现细节可能有所调整。