Lettuce创建连接过程源码分析

2018-07-03 01:03:07来源:博客园 阅读 ()

新老客户大回馈,云服务器低至5折

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

   下面看看RedisClient是如何创建单机模式的异步连接的, 首先从RedisClient中的connectAsync看起,在该方法中并没有什么特别的地方,在对RedisURI进行非空校验后就直接调用了内部方法

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

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

  在内部方法中首先通过一个异步方式创建连接,在从ConnectionFuture中获取连接

   /**
     * 获取单机连接
     */
    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);
    }

  可以看到在创建 StatefulRedisConnectionImpl实例的时候实际上是创建了多种方式连接,异步连接,同步连接响应式连接

/**
     * 初始化一个新的连接
     */
    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);
    }

  

  

标签:

版权申明:本站文章部分自网络,如有侵权,请联系:west999com@outlook.com
特别注意:本站所有转载文章言论不代表本站观点,本站所提供的摄影照片,插画,设计作品,如需使用,请与原作者联系,版权归原作者所有

上一篇:Lettuce同步命令源码分析

下一篇:架构师成长之路-基于android fragment通信的面向对象的万能接口