spring-data-redis 连接泄漏，我 TM 人傻了

喜欢ヅ旅行 2022-09-09 06:17 5阅读 0赞

> 本系列是 **我TM人傻了** 系列第四期\[捂脸\]，往期精彩回顾：
> 
>  *  [升级到Spring 5.3.x之后，GC次数急剧增加，我TM人傻了][Spring 5.3.x_GC_TM]
>  *  [这个大表走索引字段查询的 SQL 怎么就成全扫描了，我TM人傻了][SQL _TM]
>  *  [获取异常信息里再出异常就找不到日志了，我TM人傻了][TM]

![image][]

> 本文基于 Spring Data Redis 2.4.9

最近线上又出事儿了，新上线了一个微服务系统，**上线之后就开始报各种发往这个系统的请求超时，这是咋回事呢**？

![image][image 1]

还是经典的**通过 JFR 去定位（可以参考我的其他系列文章，经常用到 JFR）**，对于**历史某些请求响应慢**，我一般按照如下流程去看：

1.  是否有 STW（Stop-the-world，参考我的另一篇文章：[JVM相关 - SafePoint 与 Stop The World 全解][JVM_ - SafePoint _ Stop The World]）:
2.  是否有 GC 导致的长时间 STW
3.  是否有其他原因导致进程所有线程进入 safepoint 导致 STW
4.  是否 IO 花了太长时间，例如调用其他微服务，访问各种存储（硬盘，数据库，缓存等等）
5.  是否在某些锁上面阻塞太长时间？
6.  是否 CPU 占用过高，哪些线程导致的？

通过 JFR 发现是很多 HTTP 线程在一个锁上面阻塞了，**这个锁是从 Redis 连接池获取连接的锁**。我们的项目使用的 spring-data-redis，底层客户端使用 lettuce。为何会阻塞在这里呢？经过分析，我发现 spring-data-redis 存在**连接泄漏的问题**。

![image][image 2]

我们先来简单介绍下 Lettuce，简单来说 Lettuce 就是使用 Project Reactor + Netty 实现的 Redis 非阻塞响应式客户端。spring-data-redis 是针对 Redis 操作的统一封装。我们项目使用的是 spring-data-redis + Lettuce 的组合。

为了和大家尽量说明白问题的原因，这里先将 spring-data-redis + lettuce API 结构简单介绍下。

首先 lettuce 官方，**是不推荐使用连接池的，但是官方没有说，这是什么情况下的决定**。这里先放上结论：

*  **如果你的项目中，使用的 spring-data-redis + lettuce，并且使用的都是 Redis 简单命令，没有使用 Redis 事务，Pipeline 等等，那么不使用连接池，是最好的**（并且你没有关闭 Lettuce 连接共享，这个默认是开启的）。
 *  **如果你的项目中，大量使用了 Redis 事务，那么最好还是使用连接池**
 *  **其实更准确地说**，如果你使用了大量会触发 `execute(SessionCallback)` 的命令，最好使用连接池，如果你使用的都是 `execute(RedisCallback)` 的命令，就不太有必要使用连接池了。如果大量使用 Pipeline，最好还是使用连接池。

接下来介绍下 spring-data-redis 的 API 原理。在我们的项目中，主要使用 spring-data-redis 的两个核心 API，即同步的 `RedisTemplate` 和异步的 `ReactiveRedisTemplate`。我们这里主要以同步的 `RedisTemplate` 为例子，说明原理。`ReactiveRedisTemplate` 其实就是做了异步封装，Lettuce 本身就是异步客户端，所以 `ReactiveRedisTemplate` 其实实现更简单。

`RedisTemplate` 的一切 Redis 操作，最终都会被封装成两种操作对象，**一是 `RedisCallback<T>`**：

public interface RedisCallback<T> {
    	@Nullable
    	T doInRedis(RedisConnection connection) throws DataAccessException;
    }

是一个 Functional Interface，入参是 `RedisConnection`，可以通过使用 `RedisConnection` 操作 Redis。可以是若干个 Redis 操作的集合。大部分 `RedisTemplate` 的简单 Redis 操作都是通过这个实现的。例如 Get 请求的源码实现就是：

//在 RedisCallback 的基础上增加统一反序列化的操作
    abstract class ValueDeserializingRedisCallback implements RedisCallback<V> {
    	private Object key;
    
    	public ValueDeserializingRedisCallback(Object key) {
    		this.key = key;
    	}
    
    	public final V doInRedis(RedisConnection connection) {
    		byte[] result = inRedis(rawKey(key), connection);
    		return deserializeValue(result);
    	}
    
    	@Nullable
    	protected abstract byte[] inRedis(byte[] rawKey, RedisConnection connection);
    }
    
    //Redis Get 命令的实现
    
    public V get(Object key) {
    
    	return execute(new ValueDeserializingRedisCallback(key) {
    
    		@Override
    		protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
    		    //使用 connection 执行 get 命令
    			return connection.get(rawKey);
    		}
    	}, true);
    }

**另一种是`SessionCallback<T>`**：

public interface SessionCallback<T> {
    
    	@Nullable
    	<K, V> T execute(RedisOperations<K, V> operations) throws DataAccessException;
    }

`SessionCallback`也是一个 Functional Interface，方法体也是可以放若干个命令。顾名思义，即在这个方法中的所有命令，**都是会共享同一个会话，即使用的 Redis 连接是同一个并且不能被共享的**。一般如果使用 Redis 事务则会使用这个实现。

`RedisTemplate` 的 API 主要是以下这几个，所有的命令底层实现都是这几个 API：

*  `execute(RedisCallback<?> action)` 和 `executePipelined(final SessionCallback<?> session)`：执行一系列 Redis 命令，是所有方法的基础，里面使用的**连接资源会在执行后自动释放**。
 *  `executePipelined(RedisCallback<?> action)` 和 `executePipelined(final SessionCallback<?> session)`：使用 PipeLine 执行一系列命令，**连接资源会在执行后自动释放**。
 *  `executeWithStickyConnection(RedisCallback<T> callback)`：执行一系列 Redis 命令，**连接资源不会自动释放**，各种 Scan 命令就是通过这个方法实现的，因为 Scan 命令会返回一个 Cursor，这个 Cursor 需要保持连接（会话），同时交给用户决定什么时候关闭。

![image][image 3]

通过源码我们可以发现，`RedisTemplate` 的三个 API 在实际应用的时候，经常会发生互相嵌套递归的情况。

例如如下这种：

redisTemplate.executePipelined(new RedisCallback<Object>() {
        @Override
        public Object doInRedis(RedisConnection connection) throws DataAccessException {
            orders.forEach(order -> {
                connection.hashCommands().hSet(orderKey.getBytes(), order.getId().getBytes(), JSON.toJSONBytes(order));
            });
            return null;
        }
    });

和

redisTemplate.executePipelined(new RedisCallback<Object>() {
        @Override
        public Object doInRedis(RedisConnection connection) throws DataAccessException {
            orders.forEach(order -> {
                redisTemplate.opsForHash().put(orderKey, order.getId(), JSON.toJSONString(order));
            });
            return null;
        }
    });

是等价的。`redisTemplate.opsForHash().put()`其实调用的是 `execute(RedisCallback)` 方法，这种就是 `executePipelined` 与 `execute(RedisCallback)` 嵌套，由此我们可以组合出各种复杂的情况，但是里面使用的连接是怎么维护的呢？

其实这几个方法获取连接的时候，使用的都是：`RedisConnectionUtils.doGetConnection` 方法，去获取连接并执行命令。对于 Lettuce 客户端，获取的是一个 `org.springframework.data.redis.connection.lettuce.LettuceConnection`. 这个连接封装包含两个实际 Lettuce Redis 连接，分别是：

private final @Nullable StatefulConnection<byte[], byte[]> asyncSharedConn;
    
    private @Nullable StatefulConnection<byte[], byte[]> asyncDedicatedConn;

*  **asyncSharedConn**：可以为空，如果开启了连接共享，则不为空，默认是开启的；所有 LettuceConnection 共享的 Redis 连接，对于每个 LettuceConnection 实际上都是同一个连接；用于执行简单命令，因为 Netty 客户端与 Redis 的单处理线程特性，共享同一个连接也是很快的。如果没开启连接共享，则这个字段为空，使用 asyncDedicatedConn 执行命令。
 *  **asyncDedicatedConn**：私有连接，如果需要保持会话，执行事务，以及 Pipeline 命令，固定连接，则必须使用这个 asyncDedicatedConn 执行 Redis 命令。

我们通过一个简单例子来看一下执行流程，首先是一个简单命令：`redisTemplate.opsForValue().get("test")`，根据之前的源码分析，我们知道，底层其实就是 `execute(RedisCallback)`，流程是：

![image][image 4]

可以看出，如果使用的是 `RedisCallback`，那么其实不需要绑定连接，不涉及事务。Redis 连接会在回调内返回。需要注意的是，如果是调用 `executePipelined(RedisCallback)`，**需要使用回调的连接进行 Redis 调用，不能直接使用 `redisTemplate` 调用，否则 pipeline 不生效**：

**Pipeline 生效**：

List<Object> objects = redisTemplate.executePipelined(new RedisCallback<Object>() {
        @Override
        public Object doInRedis(RedisConnection connection) throws DataAccessException {
            connection.get("test".getBytes());
            connection.get("test2".getBytes());
            return null;
        }
    });

**Pipeline 不生效**：

List<Object> objects = redisTemplate.executePipelined(new RedisCallback<Object>() {
        @Override
        public Object doInRedis(RedisConnection connection) throws DataAccessException {
            redisTemplate.opsForValue().get("test");
            redisTemplate.opsForValue().get("test2");
            return null;
        }
    });

然后，我们尝试将其加入事务中，由于我们的目的不是真的测试事务，只是为了演示问题，所以，仅仅是用 `SessionCallback` 将 GET 命令包装起来：

redisTemplate.execute(new SessionCallback<Object>() {
        @Override
        public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
            return operations.opsForValue().get("test");
        }
    });

这里最大的区别就是，外层获取连接的时候，这次是 bind = true 即将连接与当前线程绑定，用于保持会话连接。外层流程如下：

![image][image 5]

里面的 `SessionCallback` 其实就是 `redisTemplate.opsForValue().get("test")`，**使用的是共享的连接，而不是独占的连接，因为我们这里还没开启事务**（即执行 multi 命令），如果开启了事务使用的就是独占的连接，流程如下：  
![image][image 6]

由于 `SessionCallback` 需要保持连接，所以流程有很大变化，首先需要绑定连接，其实就是获取连接放入 ThreadLocal 中。同时，针对 LettuceConnection 进行了封装，我们主要关注这个封装有一个引用计数的变量。每嵌套一次 `execute` 就会将这个计数 + 1，执行完之后，就会将这个计数 -1， 同时每次 `execute` 结束的时候都会检查这个引用计数，如果**引用计数归零，就会调用 `LettuceConnection.close()`**。

接下来再来看，如果是 `executePipelined(SessionCallback)` 会怎么样：

List<Object> objects = redisTemplate.executePipelined(new SessionCallback<Object>() {
        @Override
        public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
            operations.opsForValue().get("test");
            return null;
        }
    });

其实与第二个例子在流程上的主要区别在于，**使用的连接不是共享连接，而是直接是独占的连接**。

![image][image 7]

最后我们再来看一个例子，如果是在 `execute(RedisCallback)` 中执行基于 `executeWithStickyConnection(RedisCallback<T> callback)` 的命令会怎么样，各种 SCAN 就是基于 `executeWithStickyConnection(RedisCallback<T> callback)` 的，例如：

redisTemplate.execute(new SessionCallback<Object>() {
        @Override
        public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
            Cursor<Map.Entry<Object, Object>> scan = operations.opsForHash().scan((K) "key".getBytes(), ScanOptions.scanOptions().match("*").count(1000).build());
            //scan 最后一定要关闭，这里采用 try-with-resource
            try (scan) {
                
            } catch (IOException e) {
                e.printStackTrace();
            }
            return null;
        }
    });

这里 Session callback 的流程，如下图所示，因为处于 SessionCallback，所以 `executeWithStickyConnection` 会发现当前绑定了连接，于是标记 + 1，但是并不会标记 - 1，因为 `executeWithStickyConnection` 可以将资源暴露到外部，例如这里的 Cursor，需要外部手动关闭。  
![image][image 8]

![image][image 9]

在这个例子中，会发生连接泄漏，首先执行：

这样呢，**LettuceConnection 会和当前线程绑定**，并且在结束时，**引用计数不为零，而是 1**。并且 cursor 关闭时，会调用 LettuceConnection 的 close。但是 LettuceConnection 的 close 的实现，其实**只是标记状态，并且把独占的连接 `asyncDedicatedConn` 关闭**，由于当前没有使用到独占的连接，所以为空，不需要关闭；如下面源码所示：

[LettuceConnection][]：

@Override
    public void close() throws DataAccessException {
    	super.close();
    
    	if (isClosed) {
    		return;
    	}
    
    	isClosed = true;
    
    	if (asyncDedicatedConn != null) {
    		try {
    			if (customizedDatabaseIndex()) {
    				potentiallySelectDatabase(defaultDbIndex);
    			}
    			connectionProvider.release(asyncDedicatedConn);
    		} catch (RuntimeException ex) {
    			throw convertLettuceAccessException(ex);
    		}
    	}
    
    	if (subscription != null) {
    		if (subscription.isAlive()) {
    			subscription.doClose();
    		}
    		subscription = null;
    	}
    
    	this.dbIndex = defaultDbIndex;
    }

之后我们继续执行一个 Pipeline 命令：

List<Object> objects = redisTemplate.executePipelined(new RedisCallback<Object>() {
        @Override
        public Object doInRedis(RedisConnection connection) throws DataAccessException {
            connection.get("test".getBytes());
            redisTemplate.opsForValue().get("test");
            return null;
        }
    });

这时候由于连接已经绑定到当前线程，同时同上上一节分析我们知道第一步解开释放这个绑定，但是调用了 `LettuceConnection` 的 close。执行这个代码，会创建一个独占连接，并且，由于计数不能归零，导致连接一直与当前线程绑定，这样，**这个独占连接一直不会关闭（如果有连接池的话，就是一直不返回连接池）**

即使后面我们手动关闭这个链接，但是根据源码，由于状态 isClosed 已经是 true，还是不能将独占链接关闭。这样，就会造成**连接泄漏**。

针对这个 Bug，我已经向 spring-data-redis 一个 Issue：[Lettuce Connection Leak while using execute(SessionCallback) and executeWithStickyConnection in same thread by random turn][Lettuce Connection Leak while using execute_SessionCallback_ and executeWithStickyConnection in same thread by random turn]

![image][image 10]

*  尽量避免使用 `SessionCallback`，尽量仅在需要使用 Redis 事务的时候，使用 `SessionCallback`。
 *  使用 `SessionCallback` 的函数单独封装，将事务相关的命令单独放在一起，并且外层尽量避免再继续套 `RedisTemplate` 的 `execute` 相关函数。

> **微信搜索“我的编程喵”关注公众号，每日一刷，轻松提升技术，斩获各种offer**：

![3f58a37c57d531d20ebc9a795bc54dcf.gif][]

[Spring 5.3.x_GC_TM]: https://zhuanlan.zhihu.com/p/397042565
[SQL _TM]: https://zhuanlan.zhihu.com/p/397271448
[TM]: https://blog.csdn.net/zhxdick/article/details/119595157
[image]: /images/20220829/37b36945ebe146c69f84ffd93e10caec.png
[image 1]: /images/20220829/7b731c0a8b60442797a00535becf50cc.png
[JVM_ - SafePoint _ Stop The World]: https://juejin.cn/post/6854573211968143373
[image 2]: /images/20220829/98b7c21cdc6e4b66a3b1cfe29ccd20d8.png
[image 3]: /images/20220829/1314a6c4546a4fb5bdf982261053c424.png
[image 4]: /images/20220829/4ccc26f085804a17aae53ec3ba8d8464.png
[image 5]: /images/20220829/857bdef36c7f49d1a70bc9cae1fa5bec.png
[image 6]: /images/20220829/a610852e6dad476d9cf879e1f7cc6159.png
[image 7]: /images/20220829/be3c5f6efd3f453c9c636429eae6449e.png
[image 8]: /images/20220829/d2ffb99e8d614a7bb009a0c607a193c4.png
[image 9]: /images/20220829/49e104c4130d4a36934264b8639124d8.png
[LettuceConnection]: https://github.com/spring-projects/spring-data-redis/blob/main/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java
[Lettuce Connection Leak while using execute_SessionCallback_ and executeWithStickyConnection in same thread by random turn]: https://github.com/spring-projects/spring-data-redis/issues/2148
[image 10]: /images/20220829/82868817d52f4608928274f3dd744b70.png
[3f58a37c57d531d20ebc9a795bc54dcf.gif]: /images/20220829/6d38ec7153be49758ecf062139eac76d.png