当前位置: 首页 > news >正文

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);}

http://www.lryc.cn/news/506155.html

相关文章:

  • 《宇宙机器人》提示错误弹窗“找不到d3dx9_43.dll”是什么原因?“d3dx9_43.dll缺失”怎么解决?
  • 应用于项目的 C++单例基类的设计、实现与应用
  • Mongodb 启用认证
  • QT:vlc出错处理及重新播放
  • 密钥管理系统在数据安全解决方案中的重要性
  • Docker的容器编排
  • Java Web项目部署教程简单实用
  • 推送本地仓库到远程git仓库
  • 线性池学习
  • 微积分复习笔记 Calculus Volume 2 - 4.3 Separable Equations
  • 前端项目部署方法
  • Docker创建一个mongodb实例,并用springboot连接 mongodb进行读写文件
  • Android app反编译 攻与防
  • ElasticSearch 简介
  • Kerberos实验
  • Android之RecyclerView显示数据列表和网格
  • docker mysql挂载
  • 顺序表-递增有序表合并
  • 【Qt】qt安装
  • CXF WebService SpringBoot 添加拦截器,处理响应报文格式
  • vue iframe进行父子页面通信并切换URL
  • 基于Spring Boot的摄影师分享交流社区
  • Web 毕设篇-适合小白、初级入门练手的 Spring Boot Web 毕业设计项目:电影院后台管理系统(前后端源码 + 数据库 sql 脚本)
  • Linux(网络协议和管理)
  • C++ 入门第 20 天:STL 容器之无序集合与无序多重集合
  • devops-部署Harbor实现私有Docker镜像仓库
  • rebase ‘A‘ onto ‘master‘ 和 merge ‘master‘ into ‘A‘有什么区别
  • Vulhub:Jackson[漏洞复现]
  • strongswan构建测试环境
  • 前端:金额高精度处理