lettuce连接池很香，撸撸它的源代码(下）

public T borrowObject(final long borrowMaxWaitMillis) throws Exception {
        //省略部分代码
        PooledObject<T> p = null;
        // Get local copy of current config so it is consistent for entire
        // method execution
        final boolean blockWhenExhausted = getBlockWhenExhausted();
        boolean create;
        final long waitTime = System.currentTimeMillis();
        while (p == null) {
            create = false;
            p = idleObjects.pollFirst();
            if (p == null) {
                p = create();
                if (p != null) {
                    create = true;
                }
            }
            //省略部分代码
        }
        updateStatsBorrow(p, System.currentTimeMillis() - waitTime);
        return p.getObject();
    }

public void returnObject(final T obj) {
        //省略部分代码
        final int maxIdleSave = getMaxIdle();
        if (isClosed() || maxIdleSave > -1 && maxIdleSave <= idleObjects.size()) {
            try {
                destroy(p);
            } catch (final Exception e) {
                swallowException(e);
            }
        } else {
            if (getLifo()) {
                idleObjects.addFirst(p);
            } else {
                idleObjects.addLast(p);
            }
            if (isClosed()) {
                // Pool closed while object was being added to idle objects.
                // Make sure the returned object is destroyed rather than left
                // in the idle object pool (which would effectively be a leak)
                clear();
            }
        }
        updateStatsReturn(activeTime);
    }

public void close() {
  //省略部分代码
    closeAsync().join();
}
public CompletableFuture<Void> closeAsync() {
    //省略部分代码
        if (CLOSED.compareAndSet(this, ST_OPEN, ST_CLOSED)) {
            active = false;
            CompletableFuture<Void> future = channelWriter.closeAsync();
      //省略部分代码
        }
        return closeFuture;
    }

public CompletableFuture<Void> closeAsync() {
    //省略部分代码
        if (STATUS.compareAndSet(this, ST_OPEN, ST_CLOSED)) {
            Channel channel = getOpenChannel();
            if (channel != null) {
                Futures.adapt(channel.close(), closeFuture);
            } else {
                closeFuture.complete(null);
            }
        }
        return closeFuture;
    }

protected Mono<Health> doHealthCheck(Health.Builder builder) {
  //getConnection()创建一个连接
  return getConnection().flatMap((connection) -> doHealthCheck(builder, connection));
}
public LettuceReactiveRedisConnection getReactiveConnection() {
  //下面的构造函数会创建交互式连接
  return getShareNativeConnection()
      ? new LettuceReactiveRedisConnection(getSharedReactiveConnection(), reactiveConnectionProvider)
      : new LettuceReactiveRedisConnection(reactiveConnectionProvider);
}
LettuceReactiveRedisConnection(StatefulConnection<ByteBuffer, ByteBuffer> sharedConnection,
      LettuceConnectionProvider connectionProvider) {
    Assert.notNull(sharedConnection, "Shared StatefulConnection must not be null!");
    Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!");
    //调用AsyncConnect构造函数创建连接方法
    this.dedicatedConnection = new AsyncConnect(connectionProvider, StatefulConnection.class);
    this.pubSubConnection = new AsyncConnect(connectionProvider, StatefulRedisPubSubConnection.class);
    this.sharedConnection = Mono.just(sharedConnection);
  }
AsyncConnect(LettuceConnectionProvider connectionProvider, Class<? extends T> connectionType) {
      Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!");
      this.connectionProvider = connectionProvider;
      //回到了之前讲的使用connectionProvider创建连接
      Mono<T> defer = Mono.defer(() -> Mono.<T> just(connectionProvider.getConnection(connectionType)));
      this.connectionPublisher = defer.subscribeOn(Schedulers.elastic());
    }

private void ensureIdle(final int idleCount, final boolean always) throws Exception {
       //省略部分代码
        while (idleObjects.size() < idleCount) {
            final PooledObject<T> p = create();
            if (p == null) {
                // Can't create objects, no reason to think another call to
                // create will work. Give up.
                break;
            }
            if (getLifo()) {
                idleObjects.addFirst(p);
            } else {
                idleObjects.addLast(p);
            }
        }
        if (isClosed()) {
            // Pool closed while object was being added to idle objects.
            // Make sure the returned object is destroyed rather than left
            // in the idle object pool (which would effectively be a leak)
            clear();
        }
    }

protected Mono<Health> doHealthCheck(Health.Builder builder) {
  return getConnection().flatMap((connection) -> doHealthCheck(builder, connection));
}

public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType) {
  GenericObjectPool<StatefulConnection<?, ?>> pool = pools.computeIfAbsent(connectionType, poolType -> {
    return ConnectionPoolSupport.createGenericObjectPool(() -> connectionProvider.getConnection(connectionType),
        poolConfig, false);
  });
  //省略部分代码
}}

public static <T extends StatefulConnection<?, ?>> GenericObjectPool<T> createGenericObjectPool(
        Supplier<T> connectionSupplier, GenericObjectPoolConfig config, boolean wrapConnections) {
  //省略部分代码
    GenericObjectPool<T> pool = new GenericObjectPool<T>(new RedisPooledObjectFactory<T>(connectionSupplier), config) {
  //省略部分代码
    };
    poolRef.set(new ObjectPoolWrapper<>(pool));
    return pool;
}

public GenericObjectPool(final PooledObjectFactory<T> factory,
            final GenericObjectPoolConfig<T> config) {
  //省略部分代码
        setConfig(config);
    }

@Bean
RedisTemplate redisTemplate(LettuceConnectionFactory factory){
    factory.setShareNativeConnection(false);
    RedisTemplate redisTemplate = new RedisTemplate();
    redisTemplate.setConnectionFactory(factory);
    return redisTemplate;
}