聊聊lettuce的shareNativeConnection参数

序

本文主要研究一下lettuce的shareNativeConnection参数

LettuceConnectionFactory

spring-data-redis-2.0.10.RELEASE-sources.jar!/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java

public class LettuceConnectionFactoryimplements InitializingBean, DisposableBean, RedisConnectionFactory, ReactiveRedisConnectionFactory {private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new PassThroughExceptionTranslationStrategy(LettuceConverters.exceptionConverter());private final Log log = LogFactory.getLog(getClass());private final LettuceClientConfiguration clientConfiguration;private @Nullable AbstractRedisClient client;private @Nullable LettuceConnectionProvider connectionProvider;private @Nullable LettuceConnectionProvider reactiveConnectionProvider;private boolean validateConnection = false;private boolean shareNativeConnection = true;private @Nullable SharedConnection<byte[]> connection;private @Nullable SharedConnection<ByteBuffer> reactiveConnection;private @Nullable LettucePool pool;/** Synchronization monitor for the shared Connection */private final Object connectionMonitor = new Object();private boolean convertPipelineAndTxResults = true;private RedisStandaloneConfiguration standaloneConfig = new RedisStandaloneConfiguration("localhost", 6379);private @Nullable RedisSentinelConfiguration sentinelConfiguration;private @Nullable RedisClusterConfiguration clusterConfiguration;private @Nullable ClusterCommandExecutor clusterCommandExecutor;//......@Overridepublic LettuceReactiveRedisConnection getReactiveConnection() {return getShareNativeConnection()? new LettuceReactiveRedisConnection(getSharedReactiveConnection(), reactiveConnectionProvider): new LettuceReactiveRedisConnection(reactiveConnectionProvider);}@Overridepublic LettuceReactiveRedisClusterConnection getReactiveClusterConnection() {if (!isClusterAware()) {throw new InvalidDataAccessApiUsageException("Cluster is not configured!");}RedisClusterClient client = (RedisClusterClient) this.client;return getShareNativeConnection()? new LettuceReactiveRedisClusterConnection(getSharedReactiveConnection(), reactiveConnectionProvider, client): new LettuceReactiveRedisClusterConnection(reactiveConnectionProvider, client);}/*** Indicates if multiple {@link LettuceConnection}s should share a single native connection.** @return native connection shared.*/public boolean getShareNativeConnection() {return shareNativeConnection;}/*** @return the shared connection using {@link ByteBuffer} encoding for reactive API use. {@literal null} if*         {@link #getShareNativeConnection() connection sharing} is disabled.* @since 2.0.1*/@Nullableprotected StatefulConnection<ByteBuffer, ByteBuffer> getSharedReactiveConnection() {return shareNativeConnection ? getOrCreateSharedReactiveConnection().getConnection() : null;}private SharedConnection<ByteBuffer> getOrCreateSharedReactiveConnection() {synchronized (this.connectionMonitor) {if (this.reactiveConnection == null) {this.reactiveConnection = new SharedConnection<>(reactiveConnectionProvider, true);}return this.reactiveConnection;}}
}
复制代码

可以看到这里的shareNativeConnection默认为true，表示多个LettuceConnection将共享一个native connection
如果该值为true，则getReactiveConnection及getReactiveClusterConnection方法使用的是getSharedReactiveConnection
getSharedReactiveConnection在shareNativeConnection为true的时候，调用的是getOrCreateSharedReactiveConnection().getConnection()

LettuceConnectionFactory.SharedConnection

spring-data-redis-2.0.10.RELEASE-sources.jar!/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java

   /*** Wrapper for shared connections. Keeps track of the connection lifecycleThe wrapper is thread-safe as it* synchronizes concurrent calls by blocking.** @param <E> connection encoding.* @author Mark Paluch* @author Christoph Strobl* @since 2.1*/@RequiredArgsConstructorclass SharedConnection<E> {private final LettuceConnectionProvider connectionProvider;private final boolean shareNativeClusterConnection;/** Synchronization monitor for the shared Connection */private final Object connectionMonitor = new Object();private @Nullable StatefulConnection<E, E> connection;/*** Returns a valid Lettuce connection. Initializes and validates the connection if* {@link #setValidateConnection(boolean) enabled}.** @return the connection.*/@NullableStatefulConnection<E, E> getConnection() {synchronized (this.connectionMonitor) {if (this.connection == null) {this.connection = getNativeConnection();}if (getValidateConnection()) {validateConnection();}return this.connection;}}/*** Obtain a connection from the associated {@link LettuceConnectionProvider}.** @return the connection.*/@Nullableprivate StatefulConnection<E, E> getNativeConnection() {try {if (isClusterAware() && !shareNativeClusterConnection) {return null;}StatefulConnection<E, E> connection = connectionProvider.getConnection(StatefulConnection.class);if (connection instanceof StatefulRedisConnection && getDatabase() > 0) {((StatefulRedisConnection) connection).sync().select(getDatabase());}return connection;} catch (RedisException e) {throw new RedisConnectionFailureException("Unable to connect to Redis", e);}}/*** Validate the connection. Invalid connections will be closed and the connection state will be reset.*/void validateConnection() {synchronized (this.connectionMonitor) {boolean valid = false;if (connection != null && connection.isOpen()) {try {if (connection instanceof StatefulRedisConnection) {((StatefulRedisConnection) connection).sync().ping();}if (connection instanceof StatefulRedisClusterConnection) {((StatefulRedisConnection) connection).sync().ping();}valid = true;} catch (Exception e) {log.debug("Validation failed", e);}}if (!valid) {if (connection != null) {connectionProvider.release(connection);}log.warn("Validation of shared connection failed. Creating a new connection.");resetConnection();this.connection = getNativeConnection();}}}/*** Reset the underlying shared Connection, to be reinitialized on next access.*/void resetConnection() {synchronized (this.connectionMonitor) {if (this.connection != null) {this.connectionProvider.release(this.connection);}this.connection = null;}}}
复制代码

要注意这里维护了StatefulConnection，第一个为null的时候，才调用getNativeConnection去获取
另外要注意，这里的getValidateConnection，默认是false的，也就是说只要connection不为null，就不会归还，每次用同一个connection
如果开启validate的话，每次get的时候都会validate一下，而其validate方法不仅判断isOpen，还判断ping，如果超时等，则会将连接释/归还，再重新获取一次(如果使用连接池的话，则重新borrow一次)
这里的validateConnection方法有点问题，调用了两次connectionProvider.release(connection)

LettucePoolingConnectionProvider.release

spring-data-redis-2.0.10.RELEASE-sources.jar!/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java

    @Overridepublic void release(StatefulConnection<?, ?> connection) {GenericObjectPool<StatefulConnection<?, ?>> pool = poolRef.remove(connection);if (pool == null) {throw new PoolException("Returned connection " + connection+ " was either previously returned or does not belong to this connection provider");}pool.returnObject(connection);}
复制代码

第二次remove同一个connection的时候，pool为null，然后抛出PoolException，来不及执行returnObject方法

ConnectionWatchdog

lettuce-core-5.0.5.RELEASE-sources.jar!/io/lettuce/core/protocol/ConnectionWatchdog.java

/*** A netty {@link ChannelHandler} responsible for monitoring the channel and reconnecting when the connection is lost.** @author Will Glozer* @author Mark Paluch* @author Koji Lin*/
@ChannelHandler.Sharable
public class ConnectionWatchdog extends ChannelInboundHandlerAdapter {//......@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {logger.debug("{} channelInactive()", logPrefix());if (!armed) {logger.debug("{} ConnectionWatchdog not armed", logPrefix());return;}channel = null;if (listenOnChannelInactive && !reconnectionHandler.isReconnectSuspended()) {scheduleReconnect();} else {logger.debug("{} Reconnect scheduling disabled", logPrefix(), ctx);}super.channelInactive(ctx);}/*** Schedule reconnect if channel is not available/not active.*/public void scheduleReconnect() {logger.debug("{} scheduleReconnect()", logPrefix());if (!isEventLoopGroupActive()) {logger.debug("isEventLoopGroupActive() == false");return;}if (!isListenOnChannelInactive()) {logger.debug("Skip reconnect scheduling, listener disabled");return;}if ((channel == null || !channel.isActive()) && reconnectSchedulerSync.compareAndSet(false, true)) {attempts++;final int attempt = attempts;int timeout = (int) reconnectDelay.createDelay(attempt).toMillis();logger.debug("{} Reconnect attempt {}, delay {}ms", logPrefix(), attempt, timeout);this.reconnectScheduleTimeout = timer.newTimeout(it -> {reconnectScheduleTimeout = null;if (!isEventLoopGroupActive()) {logger.warn("Cannot execute scheduled reconnect timer, reconnect workers are terminated");return;}reconnectWorkers.submit(() -> {ConnectionWatchdog.this.run(attempt);return null;});}, timeout, TimeUnit.MILLISECONDS);// Set back to null when ConnectionWatchdog#run runs earlier than reconnectScheduleTimeout's assignment.if (!reconnectSchedulerSync.get()) {reconnectScheduleTimeout = null;}} else {logger.debug("{} Skipping scheduleReconnect() because I have an active channel", logPrefix());}}/*** Reconnect to the remote address that the closed channel was connected to. This creates a new {@link ChannelPipeline} with* the same handler instances contained in the old channel's pipeline.** @param attempt attempt counter** @throws Exception when reconnection fails.*/public void run(int attempt) throws Exception {reconnectSchedulerSync.set(false);reconnectScheduleTimeout = null;if (!isEventLoopGroupActive()) {logger.debug("isEventLoopGroupActive() == false");return;}if (!isListenOnChannelInactive()) {logger.debug("Skip reconnect scheduling, listener disabled");return;}if (isReconnectSuspended()) {logger.debug("Skip reconnect scheduling, reconnect is suspended");return;}boolean shouldLog = shouldLog();InternalLogLevel infoLevel = InternalLogLevel.INFO;InternalLogLevel warnLevel = InternalLogLevel.WARN;if (shouldLog) {lastReconnectionLogging = System.currentTimeMillis();} else {warnLevel = InternalLogLevel.DEBUG;infoLevel = InternalLogLevel.DEBUG;}InternalLogLevel warnLevelToUse = warnLevel;try {reconnectionListener.onReconnect(new ConnectionEvents.Reconnect(attempt));logger.log(infoLevel, "Reconnecting, last destination was {}", remoteAddress);ChannelFuture future = reconnectionHandler.reconnect();future.addListener(it -> {if (it.isSuccess() || it.cause() == null) {return;}Throwable throwable = it.cause();if (ReconnectionHandler.isExecutionException(throwable)) {logger.log(warnLevelToUse, "Cannot reconnect: {}", throwable.toString());} else {logger.log(warnLevelToUse, "Cannot reconnect: {}", throwable.toString(), throwable);}if (!isReconnectSuspended()) {scheduleReconnect();}});} catch (Exception e) {logger.log(warnLevel, "Cannot reconnect: {}", e.toString());}}
}
复制代码

这个ConnectionWatchdog专门用来处理被异常close掉的channel，然后定时重连
重连采用的是ReconnectionHandler.reconnect方法

ReconnectionHandler.reconnect

lettuce-core-5.0.5.RELEASE-sources.jar!/io/lettuce/core/protocol/ReconnectionHandler.java

class ReconnectionHandler {//....../*** Initiate reconnect and return a {@link ChannelFuture} for synchronization. The resulting future either succeeds or fails.* It can be {@link ChannelFuture#cancel(boolean) canceled} to interrupt reconnection and channel initialization. A failed* {@link ChannelFuture} will close the channel.** @return reconnect {@link ChannelFuture}.*/protected ChannelFuture reconnect() {SocketAddress remoteAddress = socketAddressSupplier.get();logger.debug("Reconnecting to Redis at {}", remoteAddress);ChannelFuture connectFuture = bootstrap.connect(remoteAddress);ChannelPromise initFuture = connectFuture.channel().newPromise();initFuture.addListener((ChannelFuture it) -> {if (it.cause() != null) {connectFuture.cancel(true);close(it.channel());}});connectFuture.addListener((ChannelFuture it) -> {if (it.cause() != null) {initFuture.tryFailure(it.cause());return;}ChannelPipeline pipeline = it.channel().pipeline();RedisChannelInitializer channelInitializer = pipeline.get(RedisChannelInitializer.class);if (channelInitializer == null) {initFuture.tryFailure(new IllegalStateException("Reconnection attempt without a RedisChannelInitializer in the channel pipeline"));return;}channelInitializer.channelInitialized().whenComplete((state, throwable) -> {if (throwable != null) {if (isExecutionException(throwable)) {initFuture.tryFailure(throwable);return;}if (clientOptions.isCancelCommandsOnReconnectFailure()) {connectionFacade.reset();}if (clientOptions.isSuspendReconnectOnProtocolFailure()) {logger.error("Disabling autoReconnect due to initialization failure", throwable);setReconnectSuspended(true);}initFuture.tryFailure(throwable);return;}if (logger.isDebugEnabled()) {logger.info("Reconnected to {}, Channel {}", remoteAddress,ChannelLogDescriptor.logDescriptor(it.channel()));} else {logger.info("Reconnected to {}", remoteAddress);}initFuture.trySuccess();});});Runnable timeoutAction = () -> {initFuture.tryFailure(new TimeoutException(String.format("Reconnection attempt exceeded timeout of %d %s ",timeout, timeoutUnit)));};Timeout timeoutHandle = timer.newTimeout(it -> {if (connectFuture.isDone() && initFuture.isDone()) {return;}if (reconnectWorkers.isShutdown()) {timeoutAction.run();return;}reconnectWorkers.submit(timeoutAction);}, this.timeout, timeoutUnit);initFuture.addListener(it -> timeoutHandle.cancel());return this.currentFuture = initFuture;}
}
复制代码

ReconnectionHandler.reconnect就是基于netty来进行重新连接，连接失败
如果重连成功，则cancel掉timeoutHandle，否则就一直延时重试

小结

lettuce默认的shareNativeConnection参数为true，且validateConnection为false
如果使用线程池，则默认是borrow一次，之后就一直复用，不归还，但是对于docker pause的场景不能有效识别，一直报command timeout
对于不归还的shareNativeConnection，lettuce有个ConnectionWatchdog进行不断重连处理
如果validateConnection为true，则每次get连接的时候会进行校验，校验失败理论上则会归还到连接池，然后重新连接获取一个新的nativeConnection(建立连接不成功连接池那里会抛出org.springframework.data.redis.connection.PoolException: Could not get a resource from the pool; nested exception is io.lettuce.core.RedisConnectionException: Unable to connect to 192.168.99.100:6379)

不过由于LettuceConnectionFactory.SharedConnection的validateConnection方法在校验失败时，重复调用connectionProvider.release(connection)，导致抛出org.springframework.data.redis.connection.PoolException: Returned connection io.lettuce.core.StatefulRedisConnectionImpl@1e4ad4a was either previously returned or does not belong to this connection provider异常

doc

Connection-Pooling