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

本系列是 我TM人傻了 系列第四期[捂臉]，往期精彩回顧：

image

本文基於 Spring Data Redis 2.4.9

最近線上又出事兒了，新上線了一個微服務系統，上線之後就開始報各種發往這個系統的請求超時，這是咋回事呢？

image

還是經典的通過 JFR 去定位（可以參考我的其他系列文章，經常用到 JFR），對於歷史某些請求響應慢，我一般按照如下流程去看：

  1. 是否有 STW（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

我們先來簡單介紹下 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

通過源碼我們可以發現，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

可以看出，如果使用的是 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

裏面的 SessionCallback 其實就是 redisTemplate.opsForValue().get("test")，使用的是共享的連接，而不是獨佔的連接，因為我們這裡還沒開啟事務（即執行 multi 命令），如果開啟了事務使用的就是獨佔的連接，流程如下：
image

由於 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

最後我們再來看一個例子，如果是在 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

在這個例子中，會發生連接泄漏，首先執行：

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

這樣呢，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

image

  • 盡量避免使用 SessionCallback，盡量僅在需要使用 Redis 事務的時候，使用 SessionCallback。
  • 使用 SessionCallback 的函數單獨封裝，將事務相關的命令單獨放在一起，並且外層盡量避免再繼續套 RedisTemplate 的 execute 相關函數。

微信搜索「我的編程喵」關注公眾號，每日一刷，輕鬆提升技術，斬獲各種offer：

Tags:

VirMach 便宜 VPS

QNews

QNews