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

本文基於 Spring Data Redis 2.4.9

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

還是經典的通過 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 存在連線洩漏的問題。

我們先來簡單介紹下 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 需要保持連線（會話），同時交給使用者決定什麼時候關閉。

通過原始碼我們可以發現，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)，流程是：

可以看出，如果使用的是 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 即將連線與當前執行緒繫結，用於保持會話連線。外層流程如下：

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

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

    }

});

其實與第二個例子在流程上的主要區別在於，使用的連線不是共享連線，而是直接是獨佔的連線。

最後我們再來看一個例子，如果是在 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，需要外部手動關閉。

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

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

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

微信搜尋“我的程式設計喵”關注公眾號，每日一刷，輕鬆提升技術，斬獲各種offer：