Lettuce创建连接过程源码分析

Posted 2020-11-20 wei-zw

   Lettuce是一个高级的Redis客户端,下面通过对其创建连接过程的源码进行走读

   下面看看RedisClient是如何创建连接的

 public <K, V> ConnectionFuture<StatefulRedisConnection<K, V>> connectAsync(RedisCodec<K, V> codec, RedisURI redisURI) {

        assertNotNull(redisURI);
        return connectStandaloneAsync(codec, redisURI, redisURI.getTimeout());
    }

　　

   /**
     * 获取单机连接
     */
    private <K, V> StatefulRedisConnection<K, V> connectStandalone(RedisCodec<K, V> codec, RedisURI redisURI, Duration timeout) {
        //单机异步连接
        ConnectionFuture<StatefulRedisConnection<K, V>> future = connectStandaloneAsync(codec, redisURI, timeout);
        //获取连接
        return getConnection(future);
    }

　　下面首先看看如何创建单机异步连接

private <K, V> ConnectionFuture<StatefulRedisConnection<K, V>> connectStandaloneAsync(RedisCodec<K, V> codec,
            RedisURI redisURI, Duration timeout) {
        
        //编解码器不能为null
        assertNotNull(codec);
        //检查URI是否有效
        checkValidRedisURI(redisURI);

        logger.debug("Trying to get a Redis connection for: " + redisURI);
        //创建DefaultEndpoint
        DefaultEndpoint endpoint = new DefaultEndpoint(clientOptions);
        //创建connection,该connection是一个真正有效的connection其它的都是再此基础上进行增强
        StatefulRedisConnectionImpl<K, V> connection = newStatefulRedisConnection(endpoint, codec, timeout);
        //异步方式创建连接
        ConnectionFuture<StatefulRedisConnection<K, V>> future = connectStatefulAsync(connection, endpoint, redisURI,
                () -> new CommandHandler(clientOptions, clientResources, endpoint));
        //注册监听器，在结束时触发
        future.whenComplete((channelHandler, throwable) -> {
            //如果异常不为null则表示连接创建异常，则需要关闭连接
            if (throwable != null) {
                connection.close();
            }
        });
        //返回
        return future;
    }

　　在newStatefulRedisConnection中只是创建了连接对象,此时还不是一个可用连接

protected <K, V> StatefulRedisConnectionImpl<K, V> newStatefulRedisConnection(RedisChannelWriter channelWriter,
            RedisCodec<K, V> codec, Duration timeout) {
        return new StatefulRedisConnectionImpl<>(channelWriter, codec, timeout);
    }

　　

/**
     * 初始化一个新的连接
     */
    public StatefulRedisConnectionImpl(RedisChannelWriter writer, RedisCodec<K, V> codec, Duration timeout) {

        super(writer, timeout);

        this.codec = codec;
        //创建异步步连接
        this.async = newRedisAsyncCommandsImpl();
        //创建同步连接
        this.sync = newRedisSyncCommandsImpl();
        //创建响应式连接
        this.reactive = newRedisReactiveCommandsImpl();
    }

　　

 protected RedisAsyncCommandsImpl<K, V> newRedisAsyncCommandsImpl() {
        //使用装饰器模式对当前实例进行增强
        return new RedisAsyncCommandsImpl<>(this, codec);
    }

　　关键逻辑还是在connectionStatefulAsync中实现

private <K, V, S> ConnectionFuture<S> connectStatefulAsync(StatefulRedisConnectionImpl<K, V> connection,
            DefaultEndpoint endpoint, RedisURI redisURI, Supplier<CommandHandler> commandHandlerSupplier) {
        //connetion构造器，在Lettuce中对于构造器模式运用很多
        ConnectionBuilder connectionBuilder;
        //根据是否是SSL选择不同构造器
        if (redisURI.isSsl()) {
            SslConnectionBuilder sslConnectionBuilder = SslConnectionBuilder.sslConnectionBuilder();
            sslConnectionBuilder.ssl(redisURI);
            connectionBuilder = sslConnectionBuilder;
        } else {
            connectionBuilder = ConnectionBuilder.connectionBuilder();
        }
        //设置connection
        connectionBuilder.connection(connection);
        //设置客户端选项
        connectionBuilder.clientOptions(clientOptions);
        //设置客户端资源
        connectionBuilder.clientResources(clientResources);
        //设置命令处理器以及endpoint
        connectionBuilder.commandHandler(commandHandlerSupplier).endpoint(endpoint);
        //填充连接构造器，
        connectionBuilder(getSocketAddressSupplier(redisURI), connectionBuilder, redisURI);
        //设置频道类型，同时根据频道类型设置客户端NIO线程组
        channelType(connectionBuilder, redisURI);
         //在连接生效前是否需要ping   
        if (clientOptions.isPingBeforeActivateConnection()) {
            if (hasPassword(redisURI)) {
                connectionBuilder.enableAuthPingBeforeConnect();
            } else {
                connectionBuilder.enablePingBeforeConnect();
            }
        }
        //创建异步通道
        ConnectionFuture<RedisChannelHandler<K, V>> future = initializeChannelAsync(connectionBuilder);

        //如果客户端选项配置了pingBeforeActivateConnection同时有密码
        if (!clientOptions.isPingBeforeActivateConnection() && hasPassword(redisURI)) {

            future = future.thenApplyAsync(channelHandler -> {

                connection.async().auth(new String(redisURI.getPassword()));

                return channelHandler;
            }, clientResources.eventExecutorGroup());
        }

        if (LettuceStrings.isNotEmpty(redisURI.getClientName())) {
            future.thenApply(channelHandler -> {
                connection.setClientName(redisURI.getClientName());
                return channelHandler;
            });

        }

        if (redisURI.getDatabase() != 0) {

            future = future.thenApplyAsync(channelHandler -> {

                connection.async().select(redisURI.getDatabase());

                return channelHandler;
            }, clientResources.eventExecutorGroup());
        }

        return future.thenApply(channelHandler -> (S) connection);
    }

　　在connectionBuilder方法中创建了Netty的客户端Bootstrap

protected void connectionBuilder(Supplier<SocketAddress> socketAddressSupplier, ConnectionBuilder connectionBuilder,
            RedisURI redisURI) {
       //创建Bootstrap netty启动器
        Bootstrap redisBootstrap = new Bootstrap();
        //设置channel选项
        redisBootstrap.option(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, 32 * 1024);
        redisBootstrap.option(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, 8 * 1024);

        redisBootstrap.option(ChannelOption.ALLOCATOR, BUF_ALLOCATOR);
        //获取套接字选项
        SocketOptions socketOptions = getOptions().getSocketOptions();
        //设置连接超时时间
        redisBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS,
                Math.toIntExact(socketOptions.getConnectTimeout().toMillis()));
        //如果redisURI中没有socket选择参数则根据clientresouce设置
        if (LettuceStrings.isEmpty(redisURI.getSocket())) {
            //是否保持长连接
            redisBootstrap.option(ChannelOption.SO_KEEPALIVE, socketOptions.isKeepAlive());
            //是否要求TCP低延迟
            redisBootstrap.option(ChannelOption.TCP_NODELAY, socketOptions.isTcpNoDelay());
        }
        //设置超时时间
        connectionBuilder.timeout(redisURI.getTimeout());
        //设置密码
        connectionBuilder.password(redisURI.getPassword());
        //设置bootstrap
        connectionBuilder.bootstrap(redisBootstrap);
        connectionBuilder.channelGroup(channels).connectionEvents(connectionEvents).timer(timer);
        connectionBuilder.socketAddressSupplier(socketAddressSupplier);
    }

　　在channelType方法中设置了EeventLoopGroup

 protected void channelType(ConnectionBuilder connectionBuilder, ConnectionPoint connectionPoint) {

        LettuceAssert.notNull(connectionPoint, "ConnectionPoint must not be null");
        //设置客户端线程组，EventLoopGroup用来处理所有频道事件
        connectionBuilder.bootstrap().group(getEventLoopGroup(connectionPoint));

        if (connectionPoint.getSocket() != null) {
            NativeTransports.assertAvailable();
            connectionBuilder.bootstrap().channel(NativeTransports.domainSocketChannelClass());
        } else {
            connectionBuilder.bootstrap().channel(Transports.socketChannelClass());
        }
    }

 

 /**
     *  异步处理连接同时通过connectionBuilder初始化一个通道
     */
    @SuppressWarnings("unchecked")
    protected <K, V, T extends RedisChannelHandler<K, V>> ConnectionFuture<T> initializeChannelAsync(
            ConnectionBuilder connectionBuilder) {
        //获取socketAddress
        SocketAddress redisAddress = connectionBuilder.socketAddress();

        //如果线程池关闭则抛出异常
        if (clientResources.eventExecutorGroup().isShuttingDown()) {
            throw new IllegalStateException("Cannot connect, Event executor group is terminated.");
        }

        logger.debug("Connecting to Redis at {}", redisAddress);
        //频道准备就绪future
        CompletableFuture<Channel> channelReadyFuture = new CompletableFuture<>();

        //获取bootstrap
        Bootstrap redisBootstrap = connectionBuilder.bootstrap();
        //创建redis通道初始化器
        RedisChannelInitializer initializer = connectionBuilder.build();
        //设置netty的处理器
        redisBootstrap.handler(initializer);
        //netty自定设置处理
        clientResources.nettyCustomizer().afterBootstrapInitialized(redisBootstrap);
        CompletableFuture<Boolean> initFuture = initializer.channelInitialized();
        //连接Redis服务器，在该处才是真正和服务器创建连接
        ChannelFuture connectFuture = redisBootstrap.connect(redisAddress);
        //增加监听器
        connectFuture.addListener(future -> {
            //没有成功
            if (!future.isSuccess()) {

                logger.debug("Connecting to Redis at {}: {}", redisAddress, future.cause());
                connectionBuilder.endpoint().initialState();
                //通过准备就绪异步结果异常结束
                channelReadyFuture.completeExceptionally(future.cause());
                return;
            }
            //completableFuture特性，在future结束的时候执行
            initFuture.whenComplete((success, throwable) -> {
                //如果throwable不为null表示存在异常
                if (throwable == null) {
                    logger.debug("Connecting to Redis at {}: Success", redisAddress);
                    //获取RedisChannelHandler
                    RedisChannelHandler<?, ?> connection = connectionBuilder.connection();
                    //注册可关闭资源，在connection关闭的时候关闭可关闭资源
                    connection.registerCloseables(closeableResources, connection);
                    //频道准备就绪
                    channelReadyFuture.complete(connectFuture.channel());
                    return;
                }

                logger.debug("Connecting to Redis at {}, initialization: {}", redisAddress, throwable);
                connectionBuilder.endpoint().initialState();
                Throwable failure;

                if (throwable instanceof RedisConnectionException) {
                    failure = throwable;
                } else if (throwable instanceof TimeoutException) {
                    failure = new RedisConnectionException("Could not initialize channel within "
                            + connectionBuilder.getTimeout(), throwable);
                } else {
                    failure = throwable;
                }
                channelReadyFuture.completeExceptionally(failure);

                CompletableFuture<Boolean> response = new CompletableFuture<>();
                response.completeExceptionally(failure);

            });
        });
        //针对connectionBuilder.connection()的结果进行装饰，增加获取remoteAddress功能
        return new DefaultConnectionFuture<T>(redisAddress, channelReadyFuture.thenApply(channel -> (T) connectionBuilder
                .connection()));
    }

　　在

public RedisChannelInitializer build() {
        return new PlainChannelInitializer(pingCommandSupplier, this::buildHandlers, clientResources, timeout);
    }

　　在buildHandlers中创建了一些处理器,这些处理器都是有序的

1. 命令编码器,用户将命令编码为Redis通信协议规定的格式
2. CammandHanler  lettuce核心功能
3. ConnectionWatchDog 用于自动重连
4. ConnectionEventTriger 用于发布connection事件

 protected List<ChannelHandler> buildHandlers() {

        LettuceAssert.assertState(channelGroup != null, "ChannelGroup must be set");
        LettuceAssert.assertState(connectionEvents != null, "ConnectionEvents must be set");
        LettuceAssert.assertState(connection != null, "Connection must be set");
        LettuceAssert.assertState(clientResources != null, "ClientResources must be set");
        LettuceAssert.assertState(endpoint != null, "Endpoint must be set");

        List<ChannelHandler> handlers = new ArrayList<>();
        //设置clientOptions
        connection.setOptions(clientOptions);
        //添加频道监控，如果频道有效则将频道添加到频道组中，如果频道无效则从频道组中删除
        handlers.add(new ChannelGroupListener(channelGroup));
        //添加命令编码器
        handlers.add(new CommandEncoder());
        //添加commandHander
        handlers.add(commandHandlerSupplier.get());
        //如果设置自动重连，则设置看门狗处理器
        if (clientOptions.isAutoReconnect()) {
            handlers.add(createConnectionWatchdog());
        }
        //设置connectionEvenTrigger
        handlers.add(new ConnectionEventTrigger(connectionEvents, connection, clientResources.eventBus()));

        if (clientOptions.isAutoReconnect()) {
            handlers.add(createConnectionWatchdog());
        }

        return handlers;
    }

　　

@Override
    protected void initChannel(Channel channel) throws Exception {

        //如果pipeline中没有配置channelActivator则需要添加channelActivator处理器
        if (channel.pipeline().get("channelActivator") == null) {

            channel.pipeline().addLast("channelActivator", new RedisChannelInitializerImpl() {

                private AsyncCommand<?, ?, ?> pingCommand;

                @Override
                public CompletableFuture<Boolean> channelInitialized() {
                    return initializedFuture;
                }

                @Override
                public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                    //如果通道断开连接
                    clientResources.eventBus().publish(new DisconnectedEvent(local(ctx), remote(ctx)));
                    //如果初始化没有完成则抛出异常
                    if (!initializedFuture.isDone()) {
                        initializedFuture.completeExceptionally(new RedisConnectionException("Connection closed prematurely"));
                    }

                    initializedFuture = new CompletableFuture<>();
                    pingCommand = null;
                    super.channelInactive(ctx);
                }

                @Override
                public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {

                    if (evt instanceof ConnectionEvents.Activated) {
                        if (!initializedFuture.isDone()) {
                            initializedFuture.complete(true);
                            clientResources.eventBus().publish(new ConnectionActivatedEvent(local(ctx), remote(ctx)));
                        }
                    }
                    super.userEventTriggered(ctx, evt);
                }

                @Override
                public void channelActive(final ChannelHandlerContext ctx) throws Exception {
                    //通过事件总线发送连接事件
                    clientResources.eventBus().publish(new ConnectedEvent(local(ctx), remote(ctx)));
                    //如果ping命令提供器不是NO_PING则发送执行ping
                    if (pingCommandSupplier != NO_PING) {
                        pingCommand = pingCommandSupplier.get();
                        pingBeforeActivate(pingCommand, initializedFuture, ctx, clientResources, timeout);
                    } else {
                        super.channelActive(ctx);
                    }
                }

                @Override
                public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                    if (!initializedFuture.isDone()) {
                        initializedFuture.completeExceptionally(cause);
                    }
                    super.exceptionCaught(ctx, cause);
                }
            });
        }
        //将hanler提供器提供的的处理器添加到该频道的管道中
        for (ChannelHandler handler : handlers.get()) {
            channel.pipeline().addLast(handler);
        }
        //扩展点，用户可以对向pipline中添加自定义的channel
        clientResources.nettyCustomizer().afterChannelInitialized(channel);
    }