lettuce 默认情况下连接池参数不生效，源码分析

先说结论：

1.LettuceConnectionFactory 属性 shareNativeConnection 默认为true，要想连接池生效，该参数设置为false;
2.使用redisTemplate模版封装的pipeline没有意义，autoFlashCommands 默认为true;

spring2.0开始默认使用lettuce,lettuce和jedis一样支持使用连接池；这里不对比两款客户端的性能差异，只针对使用lettuce客户端执行命令获取连接的源码分析其逻辑;

版本说明：

spring-date-redis:2.3.9.RELEASE;

lettuce-core:5.3.7.RELEASE;

以单机版本reids说明；即：RedisStandaloneConfiguration,集群也是同样方法

以 stringRedisTemplate.opsForValue().get(key) 示例追踪说明；

点击进入get方法：

public V get(Object key) {return this.execute(new AbstractOperations<K, V>.ValueDeserializingRedisCallback(key) {protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {return connection.get(rawKey);}}, true);
}

接着进入execute方法：

@Nullable
<T> T execute(RedisCallback<T> callback, boolean exposeConnection) {return this.template.execute(callback, exposeConnection);
}

接着进入：org.springframework.data.redis.core.RedisTemplate#execute(org.springframework.data.redis.core.RedisCallback<T>, boolean, boolean)；

@Nullable
public <T> T execute(RedisCallback<T> action, boolean exposeConnection, boolean pipeline) {Assert.isTrue(this.initialized, "template not initialized; call afterPropertiesSet() before using it");Assert.notNull(action, "Callback object must not be null");RedisConnectionFactory factory = this.getRequiredConnectionFactory();RedisConnection conn = null;Object var11;try {if (this.enableTransactionSupport) {conn = RedisConnectionUtils.bindConnection(factory, this.enableTransactionSupport);} else {conn = RedisConnectionUtils.getConnection(factory);}boolean existingConnection = TransactionSynchronizationManager.hasResource(factory);RedisConnection connToUse = this.preProcessConnection(conn, existingConnection);boolean pipelineStatus = connToUse.isPipelined();if (pipeline && !pipelineStatus) {connToUse.openPipeline();}RedisConnection connToExpose = exposeConnection ? connToUse : this.createRedisConnectionProxy(connToUse);T result = action.doInRedis(connToExpose);if (pipeline && !pipelineStatus) {connToUse.closePipeline();}var11 = this.postProcessResult(result, connToUse, existingConnection);} finally {RedisConnectionUtils.releaseConnection(conn, factory, this.enableTransactionSupport);}return var11;
}

到这里后进入获取连接的入口：conn = RedisConnectionUtils.getConnection(factory);

点进去到：org.springframework.data.redis.core.RedisConnectionUtils#doGetConnection

public static RedisConnection doGetConnection(RedisConnectionFactory factory, boolean allowCreate, boolean bind, boolean transactionSupport) {Assert.notNull(factory, "No RedisConnectionFactory specified");RedisConnectionUtils.RedisConnectionHolder connHolder = (RedisConnectionUtils.RedisConnectionHolder)TransactionSynchronizationManager.getResource(factory);if (connHolder != null) {if (transactionSupport) {potentiallyRegisterTransactionSynchronisation(connHolder, factory);}return connHolder.getConnection();} else if (!allowCreate) {throw new IllegalArgumentException("No connection found and allowCreate = false");} else {if (log.isDebugEnabled()) {log.debug("Opening RedisConnection");}RedisConnection conn = factory.getConnection();if (bind) {RedisConnection connectionToBind = conn;if (transactionSupport && isActualNonReadonlyTransactionActive()) {connectionToBind = createConnectionProxy(conn, factory);}connHolder = new RedisConnectionUtils.RedisConnectionHolder(connectionToBind);TransactionSynchronizationManager.bindResource(factory, connHolder);if (transactionSupport) {potentiallyRegisterTransactionSynchronisation(connHolder, factory);}return connHolder.getConnection();} else {return conn;}}
}

忽略事务相关逻辑，进入：factory.getConnection();

public RedisConnection getConnection() {if (this.isClusterAware()) {return this.getClusterConnection();} else {LettuceConnection connection = this.doCreateLettuceConnection(this.getSharedConnection(), this.connectionProvider, this.getTimeout(), this.getDatabase());connection.setConvertPipelineAndTxResults(this.convertPipelineAndTxResults);return connection;}
}

先看：this.getSharedConnection()；核心逻辑this.shareNativeConnection 标志位判断，默认为true;

org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory#getSharedConnection

@Nullableprotected StatefulRedisConnection<byte[], byte[]> getSharedConnection() {return this.shareNativeConnection ? (StatefulRedisConnection)this.getOrCreateSharedConnection().getConnection() : null;}

在进入核心方法：this.getOrCreateSharedConnection().getConnection()；this.connection同步只会创建一次

private LettuceConnectionFactory.SharedConnection<byte[]> getOrCreateSharedConnection() {synchronized(this.connectionMonitor) {if (this.connection == null) {this.connection = new LettuceConnectionFactory.SharedConnection(this.connectionProvider);}return this.connection;}}

@NullableStatefulConnection<E, E> getConnection() {synchronized(this.connectionMonitor) {if (this.connection == null) {this.connection = this.getNativeConnection();}if (LettuceConnectionFactory.this.getValidateConnection()) {this.validateConnection();}return this.connection;}}

private StatefulConnection<E, E> getNativeConnection() {return this.connectionProvider.getConnection(StatefulConnection.class);}

public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType) {GenericObjectPool pool = (GenericObjectPool)this.pools.computeIfAbsent(connectionType, (poolType) -> {return ConnectionPoolSupport.createGenericObjectPool(() -> {return this.connectionProvider.getConnection(connectionType);}, this.poolConfig, false);});try {StatefulConnection<?, ?> connection = (StatefulConnection)pool.borrowObject();this.poolRef.put(connection, pool);return (StatefulConnection)connectionType.cast(connection);} catch (Exception var4) {throw new PoolException("Could not get a resource from the pool", var4);}}

这里只会从pool里获取一个链接，后面不会再获取

进入：this.doCreateLettuceConnection(this.getSharedConnection(), this.connectionProvider, this.getTimeout(), this.getDatabase())

protected LettuceConnection doCreateLettuceConnection(@Nullable StatefulRedisConnection<byte[], byte[]> sharedConnection, LettuceConnectionProvider connectionProvider, long timeout, int database) {LettuceConnection connection = new LettuceConnection(sharedConnection, connectionProvider, timeout, database);connection.setPipeliningFlushPolicy(this.pipeliningFlushPolicy);return connection;}

其中 LettuceConnection connection = new LettuceConnection(sharedConnection, connectionProvider, timeout, database); 每次会把同一个连接赋值给asyncSharedConn

LettuceConnection(@Nullable StatefulConnection<byte[], byte[]> sharedConnection, LettuceConnectionProvider connectionProvider, long timeout, int defaultDbIndex) {this.isClosed = false;this.isMulti = false;this.isPipelined = false;this.txResults = new LinkedList();this.convertPipelineAndTxResults = true;this.pipeliningFlushPolicy = LettuceConnection.PipeliningFlushPolicy.flushEachCommand();Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null.");this.asyncSharedConn = sharedConnection;this.connectionProvider = connectionProvider;this.timeout = timeout;this.defaultDbIndex = defaultDbIndex;this.dbIndex = this.defaultDbIndex;}

回到命令执行方法：org.springframework.data.redis.core.RedisTemplate#execute(org.springframework.data.redis.core.RedisCallback<T>, boolean, boolean) 

来到：T result = action.doInRedis(connToExpose);

会进到：org.springframework.data.redis.connection.DefaultStringRedisConnection#get(byte[])

this.delegate.get(key)会进入到：org.springframework.data.redis.connection.lettuce.LettuceStringCommands#get

进入this.getConnection;

org.springframework.data.redis.connection.lettuce.LettuceConnection#getConnection()

到这里就用到了之前在 LettuceConnection connection = new LettuceConnection(sharedConnection, connectionProvider, timeout, database); 方法里赋值给asyncSharedConn 的对象；

如果为null即上面shareNativeConnection为false,会走 this.getDedicatedConnection() ：org.springframework.data.redis.connection.lettuce.LettucePoolingConnectionProvider#getConnection 从连接池里获取链接

获取到链接之后会进入：io.lettuce.core.AbstractRedisAsyncCommands#dispatch(io.lettuce.core.protocol.RedisCommand<K,V,T>)

至此已经能够看出shareNativeConnection参数true和false的区别

单独看下pipeline的执行方法，

首先pipeline获取链接的方法为：org.springframework.data.redis.connection.lettuce.LettuceConnection#getOrCreateDedicatedConnection

private StatefulConnection<byte[], byte[]> getOrCreateDedicatedConnection() {if (this.asyncDedicatedConn == null) {this.asyncDedicatedConn = this.doGetAsyncDedicatedConnection();}return this.asyncDedicatedConn;}

赋值给了asyncDedicatedConn;后面不在赋值；跟简单命令set/get使用的不是相同的链接;但是获取链接的方法少了synchronized同步关键字；猜想作者觉得pipeline使用场景并发不大不用同步；

然后找到pipeline命令执行方法，可以在此处断点验证，pipeline第一条命令执行完redis是否已经有值了：io.lettuce.core.protocol.DefaultEndpoint#write(io.lettuce.core.protocol.RedisCommand<K,V,T>)

这个属性autoFlushCommands 在对象创建时赋值，目前没有提供动态配置：

public DefaultEndpoint(ClientOptions clientOptions, ClientResources clientResources) {this.endpointId = ENDPOINT_COUNTER.incrementAndGet();this.sharedLock = new SharedLock();this.debugEnabled = logger.isDebugEnabled();this.closeFuture = new CompletableFuture();this.autoFlushCommands = true;this.inActivation = false;this.queueSize = 0;this.status = 0;LettuceAssert.notNull(clientOptions, "ClientOptions must not be null");LettuceAssert.notNull(clientOptions, "ClientResources must not be null");this.clientOptions = clientOptions;this.clientResources = clientResources;this.reliability = clientOptions.isAutoReconnect() ? DefaultEndpoint.Reliability.AT_LEAST_ONCE : DefaultEndpoint.Reliability.AT_MOST_ONCE;this.disconnectedBuffer = LettuceFactories.newConcurrentQueue(clientOptions.getRequestQueueSize());this.commandBuffer = LettuceFactories.newConcurrentQueue(clientOptions.getRequestQueueSize());this.boundedQueues = clientOptions.getRequestQueueSize() != 2147483647;this.rejectCommandsWhileDisconnected = isRejectCommand(clientOptions);}