Redis高級客戶端Lettuce詳解

作者： throwable

出處：https://www.cnblogs.com/throwable/p/11601538.html

spring-data-redis:  https://spring.io/projects/spring-data-redis#overview

lettuce: https://github.com/lettuce-io/lettuce-core/wiki/About-Lettuce

 

前提#

Lettuce是一個Redis的Java驅動包，初識她的時候是使用RedisTemplate的時候遇到點問題Debug到底層的一些源碼，發現spring-data-redis的驅動包在某個版本之后替換為Lettuce。Lettuce翻譯為生菜，沒錯，就是吃的那種生菜，所以它的Logo長這樣：

既然能被Spring生態所認可，Lettuce想必有過人之處，於是筆者花時間閱讀她的官方文檔，整理測試示例，寫下這篇文章。編寫本文時所使用的版本為Lettuce 5.1.8.RELEASE，SpringBoot 2.1.8.RELEASE，JDK [8,11]。超長警告：這篇文章斷斷續續花了兩周完成，超過4萬字.....

Lettuce簡介#

Lettuce是一個高性能基於Java編寫的Redis驅動框架，底層集成了Project Reactor提供天然的反應式編程，通信框架集成了Netty使用了非阻塞IO，5.x版本之后融合了JDK1.8的異步編程特性，在保證高性能的同時提供了十分豐富易用的API，5.1版本的新特性如下：

• 支持Redis的新增命令ZPOPMIN, ZPOPMAX, BZPOPMIN, BZPOPMAX。
• 支持通過Brave模塊跟蹤Redis命令執行。
• 支持Redis Streams。
• 支持異步的主從連接。
• 支持異步連接池。
• 新增命令最多執行一次模式（禁止自動重連）。
• 全局命令超時設置（對異步和反應式命令也有效）。
• ......等等

注意一點：Redis的版本至少需要2.6，當然越高越好，API的兼容性比較強大。

只需要引入單個依賴就可以開始愉快地使用Lettuce：

• Maven

Copy

<dependency> <groupId>io.lettuce</groupId> <artifactId>lettuce-core</artifactId> <version>5.1.8.RELEASE</version> </dependency> 

• Gradle

Copy

dependencies {
  compile 'io.lettuce:lettuce-core:5.1.8.RELEASE'
}

連接Redis#

單機、哨兵、集群模式下連接Redis需要一個統一的標准去表示連接的細節信息，在Lettuce中這個統一的標准是RedisURI。可以通過三種方式構造一個RedisURI實例：

• 定制的字符串URI語法：

Copy

RedisURI uri = RedisURI.create("redis://localhost/"); 

• 使用建造器（RedisURI.Builder）：

Copy

RedisURI uri = RedisURI.builder().withHost("localhost").withPort(6379).build(); 

• 直接通過構造函數實例化：

Copy

RedisURI uri = new RedisURI("localhost", 6379, 60, TimeUnit.SECONDS); 

定制的連接URI語法#

• 單機（前綴為redis://）

Copy

格式：redis://[password@]host[:port][/databaseNumber][?[timeout=timeout[d|h|m|s|ms|us|ns]]
完整：redis://mypassword@127.0.0.1:6379/0?timeout=10s
簡單：redis://localhost

• 單機並且使用SSL（前綴為rediss://） <== 注意后面多了個s

Copy

格式：rediss://[password@]host[:port][/databaseNumber][?[timeout=timeout[d|h|m|s|ms|us|ns]]
完整：rediss://mypassword@127.0.0.1:6379/0?timeout=10s
簡單：rediss://localhost

• 單機Unix Domain Sockets模式（前綴為redis-socket://）

Copy

格式：redis-socket://path[?[timeout=timeout[d|h|m|s|ms|us|ns]][&_database=database_]]
完整：redis-socket:///tmp/redis?timeout=10s&_database=0

• 哨兵（前綴為redis-sentinel://）

Copy

格式：redis-sentinel://[password@]host[:port][,host2[:port2]][/databaseNumber][?[timeout=timeout[d|h|m|s|ms|us|ns]]#sentinelMasterId
完整：redis-sentinel://mypassword@127.0.0.1:6379,127.0.0.1:6380/0?timeout=10s#mymaster

超時時間單位：

• d 天
• h 小時
• m 分鍾
• s 秒鍾
• ms 毫秒
• us 微秒
• ns 納秒

個人建議使用RedisURI提供的建造器，畢竟定制的URI雖然簡潔，但是比較容易出現人為錯誤。鑒於筆者沒有SSL和Unix Domain Socket的使用場景，下面不對這兩種連接方式進行列舉。

基本使用#

Lettuce使用的時候依賴於四個主要組件：

• RedisURI：連接信息。
• RedisClient：Redis客戶端，特殊地，集群連接有一個定制的RedisClusterClient。
• Connection：Redis連接，主要是StatefulConnection或者StatefulRedisConnection的子類，連接的類型主要由連接的具體方式（單機、哨兵、集群、訂閱發布等等）選定，比較重要。
• RedisCommands：Redis命令API接口，基本上覆蓋了Redis發行版本的所有命令，提供了同步（sync）、異步（async）、反應式（reative）的調用方式，對於使用者而言，會經常跟RedisCommands系列接口打交道。

一個基本使用例子如下：

Copy

@Test public void testSetGet() throws Exception { RedisURI redisUri = RedisURI.builder() // <1> 創建單機連接的連接信息 .withHost("localhost") .withPort(6379) .withTimeout(Duration.of(10, ChronoUnit.SECONDS)) .build(); RedisClient redisClient = RedisClient.create(redisUri); // <2> 創建客戶端 StatefulRedisConnection<String, String> connection = redisClient.connect(); // <3> 創建線程安全的連接 RedisCommands<String, String> redisCommands = connection.sync(); // <4> 創建同步命令 SetArgs setArgs = SetArgs.Builder.nx().ex(5); String result = redisCommands.set("name", "throwable", setArgs); Assertions.assertThat(result).isEqualToIgnoringCase("OK"); result = redisCommands.get("name"); Assertions.assertThat(result).isEqualTo("throwable"); // ... 其他操作 connection.close(); // <5> 關閉連接 redisClient.shutdown(); // <6> 關閉客戶端 } 

注意：

• <5>：關閉連接一般在應用程序停止之前操作，一個應用程序中的一個Redis驅動實例不需要太多的連接（一般情況下只需要一個連接實例就可以，如果有多個連接的需要可以考慮使用連接池，其實Redis目前處理命令的模塊是單線程，在客戶端多個連接多線程調用理論上沒有效果）。
• <6>：關閉客戶端一般應用程序停止之前操作，如果條件允許的話，基於后開先閉原則，客戶端關閉應該在連接關閉之后操作。

API#

Lettuce主要提供三種API：

• 同步（sync）：RedisCommands。
• 異步（async）：RedisAsyncCommands。
• 反應式（reactive）：RedisReactiveCommands。

先准備好一個單機Redis連接備用：

Copy

private static StatefulRedisConnection<String, String> CONNECTION; private static RedisClient CLIENT; @BeforeClass public static void beforeClass() { RedisURI redisUri = RedisURI.builder() .withHost("localhost") .withPort(6379) .withTimeout(Duration.of(10, ChronoUnit.SECONDS)) .build(); CLIENT = RedisClient.create(redisUri); CONNECTION = CLIENT.connect(); } @AfterClass public static void afterClass() throws Exception { CONNECTION.close(); CLIENT.shutdown(); } 

Redis命令API的具體實現可以直接從StatefulRedisConnection實例獲取，見其接口定義：

Copy

public interface StatefulRedisConnection<K, V> extends StatefulConnection<K, V> { boolean isMulti(); RedisCommands<K, V> sync(); RedisAsyncCommands<K, V> async(); RedisReactiveCommands<K, V> reactive(); } 

值得注意的是，在不指定編碼解碼器RedisCodec的前提下，RedisClient創建的StatefulRedisConnection實例一般是泛型實例StatefulRedisConnection<String,String>，也就是所有命令API的KEY和VALUE都是String類型，這種使用方式能滿足大部分的使用場景。當然，必要的時候可以定制編碼解碼器RedisCodec<K,V>。

同步API#

先構建RedisCommands實例：

Copy

private static RedisCommands<String, String> COMMAND; @BeforeClass public static void beforeClass() { COMMAND = CONNECTION.sync(); } 

基本使用：

Copy

@Test public void testSyncPing() throws Exception { String pong = COMMAND.ping(); Assertions.assertThat(pong).isEqualToIgnoringCase("PONG"); } @Test public void testSyncSetAndGet() throws Exception { SetArgs setArgs = SetArgs.Builder.nx().ex(5); COMMAND.set("name", "throwable", setArgs); String value = COMMAND.get("name"); log.info("Get value: {}", value); } // Get value: throwable 

同步API在所有命令調用之后會立即返回結果。如果熟悉Jedis的話，RedisCommands的用法其實和它相差不大。

異步API#

先構建RedisAsyncCommands實例：

Copy

private static RedisAsyncCommands<String, String> ASYNC_COMMAND; @BeforeClass public static void beforeClass() { ASYNC_COMMAND = CONNECTION.async(); } 

基本使用：

Copy

@Test public void testAsyncPing() throws Exception { RedisFuture<String> redisFuture = ASYNC_COMMAND.ping(); log.info("Ping result:{}", redisFuture.get()); } // Ping result:PONG 

RedisAsyncCommands所有方法執行返回結果都是RedisFuture實例，而RedisFuture接口的定義如下：

Copy

public interface RedisFuture<V> extends CompletionStage<V>, Future<V> { String getError(); boolean await(long timeout, TimeUnit unit) throws InterruptedException; } 

也就是，RedisFuture可以無縫使用Future或者JDK1.8中引入的CompletableFuture提供的方法。舉個例子：

Copy

@Test public void testAsyncSetAndGet1() throws Exception { SetArgs setArgs = SetArgs.Builder.nx().ex(5); RedisFuture<String> future = ASYNC_COMMAND.set("name", "throwable", setArgs); // CompletableFuture#thenAccept() future.thenAccept(value -> log.info("Set命令返回:{}", value)); // Future#get() future.get(); } // Set命令返回:OK @Test public void testAsyncSetAndGet2() throws Exception { SetArgs setArgs = SetArgs.Builder.nx().ex(5); CompletableFuture<Void> result = (CompletableFuture<Void>) ASYNC_COMMAND.set("name", "throwable", setArgs) .thenAcceptBoth(ASYNC_COMMAND.get("name"), (s, g) -> { log.info("Set命令返回:{}", s); log.info("Get命令返回:{}", g); }); result.get(); } // Set命令返回:OK // Get命令返回:throwable 

如果能熟練使用CompletableFuture和函數式編程技巧，可以組合多個RedisFuture完成一些列復雜的操作。

反應式API#

Lettuce引入的反應式編程框架是Project Reactor，如果沒有反應式編程經驗可以先自行了解一下Project Reactor。

構建RedisReactiveCommands實例：

Copy

private static RedisReactiveCommands<String, String> REACTIVE_COMMAND; @BeforeClass public static void beforeClass() { REACTIVE_COMMAND = CONNECTION.reactive(); } 

根據Project Reactor，RedisReactiveCommands的方法如果返回的結果只包含0或1個元素，那么返回值類型是Mono，如果返回的結果包含0到N（N大於0）個元素，那么返回值是Flux。舉個例子：

Copy

@Test public void testReactivePing() throws Exception { Mono<String> ping = REACTIVE_COMMAND.ping(); ping.subscribe(v -> log.info("Ping result:{}", v)); Thread.sleep(1000); } // Ping result:PONG @Test public void testReactiveSetAndGet() throws Exception { SetArgs setArgs = SetArgs.Builder.nx().ex(5); REACTIVE_COMMAND.set("name", "throwable", setArgs).block(); REACTIVE_COMMAND.get("name").subscribe(value -> log.info("Get命令返回:{}", value)); Thread.sleep(1000); } // Get命令返回:throwable @Test public void testReactiveSet() throws Exception { REACTIVE_COMMAND.sadd("food", "bread", "meat", "fish").block(); Flux<String> flux = REACTIVE_COMMAND.smembers("food"); flux.subscribe(log::info); REACTIVE_COMMAND.srem("food", "bread", "meat", "fish").block(); Thread.sleep(1000); } // meat // bread // fish 

舉個更加復雜的例子，包含了事務、函數轉換等：

Copy

@Test public void testReactiveFunctional() throws Exception { REACTIVE_COMMAND.multi().doOnSuccess(r -> { REACTIVE_COMMAND.set("counter", "1").doOnNext(log::info).subscribe(); REACTIVE_COMMAND.incr("counter").doOnNext(c -> log.info(String.valueOf(c))).subscribe(); }).flatMap(s -> REACTIVE_COMMAND.exec()) .doOnNext(transactionResult -> log.info("Discarded:{}", transactionResult.wasDiscarded())) .subscribe(); Thread.sleep(1000); } // OK // 2 // Discarded:false 

這個方法開啟一個事務，先把counter設置為1，再將counter自增1。

發布和訂閱#

非集群模式下的發布訂閱依賴於定制的連接StatefulRedisPubSubConnection，集群模式下的發布訂閱依賴於定制的連接StatefulRedisClusterPubSubConnection，兩者分別來源於RedisClient#connectPubSub()系列方法和RedisClusterClient#connectPubSub()：

• 非集群模式：

Copy

// 可能是單機、普通主從、哨兵等非集群模式的客戶端 RedisClient client = ... StatefulRedisPubSubConnection<String, String> connection = client.connectPubSub(); connection.addListener(new RedisPubSubListener<String, String>() { ... }); // 同步命令 RedisPubSubCommands<String, String> sync = connection.sync(); sync.subscribe("channel"); // 異步命令 RedisPubSubAsyncCommands<String, String> async = connection.async(); RedisFuture<Void> future = async.subscribe("channel"); // 反應式命令 RedisPubSubReactiveCommands<String, String> reactive = connection.reactive(); reactive.subscribe("channel").subscribe(); reactive.observeChannels().doOnNext(patternMessage -> {...}).subscribe() 

• 集群模式：

Copy

// 使用方式其實和非集群模式基本一致 RedisClusterClient clusterClient = ... StatefulRedisClusterPubSubConnection<String, String> connection = clusterClient.connectPubSub(); connection.addListener(new RedisPubSubListener<String, String>() { ... }); RedisPubSubCommands<String, String> sync = connection.sync(); sync.subscribe("channel"); // ... 

這里用單機同步命令的模式舉一個Redis鍵空間通知（Redis Keyspace Notifications）的例子：

Copy

@Test public void testSyncKeyspaceNotification() throws Exception { RedisURI redisUri = RedisURI.builder() .withHost("localhost") .withPort(6379) // 注意這里只能是0號庫 .withDatabase(0) .withTimeout(Duration.of(10, ChronoUnit.SECONDS)) .build(); RedisClient redisClient = RedisClient.create(redisUri); StatefulRedisConnection<String, String> redisConnection = redisClient.connect(); RedisCommands<String, String> redisCommands = redisConnection.sync(); // 只接收鍵過期的事件 redisCommands.configSet("notify-keyspace-events", "Ex"); StatefulRedisPubSubConnection<String, String> connection = redisClient.connectPubSub(); connection.addListener(new RedisPubSubAdapter<>() { @Override public void psubscribed(String pattern, long count) { log.info("pattern:{},count:{}", pattern, count); } @Override public void message(String pattern, String channel, String message) { log.info("pattern:{},channel:{},message:{}", pattern, channel, message); } }); RedisPubSubCommands<String, String> commands = connection.sync(); commands.psubscribe("__keyevent@0__:expired"); redisCommands.setex("name", 2, "throwable"); Thread.sleep(10000); redisConnection.close(); connection.close(); redisClient.shutdown(); } // pattern:__keyevent@0__:expired,count:1 // pattern:__keyevent@0__:expired,channel:__keyevent@0__:expired,message:name 

實際上，在實現RedisPubSubListener的時候可以單獨抽離，盡量不要設計成匿名內部類的形式。

事務和批量命令執行#

事務相關的命令就是WATCH、UNWATCH、EXEC、MULTI和DISCARD，在RedisCommands系列接口中有對應的方法。舉個例子：

Copy

// 同步模式 @Test public void testSyncMulti() throws Exception { COMMAND.multi(); COMMAND.setex("name-1", 2, "throwable"); COMMAND.setex("name-2", 2, "doge"); TransactionResult result = COMMAND.exec(); int index = 0; for (Object r : result) { log.info("Result-{}:{}", index, r); index++; } } // Result-0:OK // Result-1:OK 

Redis的Pipeline也就是管道機制可以理解為把多個命令打包在一次請求發送到Redis服務端，然后Redis服務端把所有的響應結果打包好一次性返回，從而節省不必要的網絡資源（最主要是減少網絡請求次數）。Redis對於Pipeline機制如何實現並沒有明確的規定，也沒有提供特殊的命令支持Pipeline機制。Jedis中底層采用BIO（阻塞IO）通訊，所以它的做法是客戶端緩存將要發送的命令，最后需要觸發然后同步發送一個巨大的命令列表包，再接收和解析一個巨大的響應列表包。Pipeline在Lettuce中對使用者是透明的，由於底層的通訊框架是Netty，所以網絡通訊層面的優化Lettuce不需要過多干預，換言之可以這樣理解：Netty幫Lettuce從底層實現了Redis的Pipeline機制。但是，Lettuce的異步API也提供了手動Flush的方法：

Copy

@Test public void testAsyncManualFlush() { // 取消自動flush ASYNC_COMMAND.setAutoFlushCommands(false); List<RedisFuture<?>> redisFutures = Lists.newArrayList(); int count = 5000; for (int i = 0; i < count; i++) { String key = "key-" + (i + 1); String value = "value-" + (i + 1); redisFutures.add(ASYNC_COMMAND.set(key, value)); redisFutures.add(ASYNC_COMMAND.expire(key, 2)); } long start = System.currentTimeMillis(); ASYNC_COMMAND.flushCommands(); boolean result = LettuceFutures.awaitAll(10, TimeUnit.SECONDS, redisFutures.toArray(new RedisFuture[0])); Assertions.assertThat(result).isTrue(); log.info("Lettuce cost:{} ms", System.currentTimeMillis() - start); } // Lettuce cost:1302 ms 

上面只是從文檔看到的一些理論術語，但是現實是骨感的，對比了下Jedis的Pipeline提供的方法，發現了Jedis的Pipeline執行耗時比較低：

Copy

@Test public void testJedisPipeline() throws Exception { Jedis jedis = new Jedis(); Pipeline pipeline = jedis.pipelined(); int count = 5000; for (int i = 0; i < count; i++) { String key = "key-" + (i + 1); String value = "value-" + (i + 1); pipeline.set(key, value); pipeline.expire(key, 2); } long start = System.currentTimeMillis(); pipeline.syncAndReturnAll(); log.info("Jedis cost:{} ms", System.currentTimeMillis() - start); } // Jedis cost:9 ms 

個人猜測Lettuce可能底層並非合並所有命令一次發送（甚至可能是單條發送），具體可能需要抓包才能定位。依此來看，如果真的有大量執行Redis命令的場景，不妨可以使用Jedis的Pipeline。

注意：由上面的測試推斷RedisTemplate的executePipelined()方法是假的Pipeline執行方法，使用RedisTemplate的時候請務必注意這一點。

Lua腳本執行#

Lettuce中執行Redis的Lua命令的同步接口如下：

Copy

public interface RedisScriptingCommands<K, V> { <T> T eval(String var1, ScriptOutputType var2, K... var3); <T> T eval(String var1, ScriptOutputType var2, K[] var3, V... var4); <T> T evalsha(String var1, ScriptOutputType var2, K... var3); <T> T evalsha(String var1, ScriptOutputType var2, K[] var3, V... var4); List<Boolean> scriptExists(String... var1); String scriptFlush(); String scriptKill(); String scriptLoad(V var1); String digest(V var1); } 

異步和反應式的接口方法定義差不多，不同的地方就是返回值類型，一般我們常用的是eval()、evalsha()和scriptLoad()方法。舉個簡單的例子：

Copy

private static RedisCommands<String, String> COMMANDS; private static String RAW_LUA = "local key = KEYS[1]\n" + "local value = ARGV[1]\n" + "local timeout = ARGV[2]\n" + "redis.call('SETEX', key, tonumber(timeout), value)\n" + "local result = redis.call('GET', key)\n" + "return result;"; private static AtomicReference<String> LUA_SHA = new AtomicReference<>(); @Test public void testLua() throws Exception { LUA_SHA.compareAndSet(null, COMMANDS.scriptLoad(RAW_LUA)); String[] keys = new String[]{"name"}; String[] args = new String[]{"throwable", "5000"}; String result = COMMANDS.evalsha(LUA_SHA.get(), ScriptOutputType.VALUE, keys, args); log.info("Get value:{}", result); } // Get value:throwable 

高可用和分片#

為了Redis的高可用，一般會采用普通主從（Master/Replica，這里筆者稱為普通主從模式，也就是僅僅做了主從復制，故障需要手動切換）、哨兵和集群。普通主從模式可以獨立運行，也可以配合哨兵運行，只是哨兵提供自動故障轉移和主節點提升功能。普通主從和哨兵都可以使用MasterSlave，通過入參包括RedisClient、編碼解碼器以及一個或者多個RedisURI獲取對應的Connection實例。

這里注意一點，MasterSlave中提供的方法如果只要求傳入一個RedisURI實例，那么Lettuce會進行拓撲發現機制，自動獲取Redis主從節點信息；如果要求傳入一個RedisURI集合，那么對於普通主從模式來說所有節點信息是靜態的，不會進行發現和更新。

拓撲發現的規則如下：

• 對於普通主從（Master/Replica）模式，不需要感知RedisURI指向從節點還是主節點，只會進行一次性的拓撲查找所有節點信息，此后節點信息會保存在靜態緩存中，不會更新。
• 對於哨兵模式，會訂閱所有哨兵實例並偵聽訂閱/發布消息以觸發拓撲刷新機制，更新緩存的節點信息，也就是哨兵天然就是動態發現節點信息，不支持靜態配置。

拓撲發現機制的提供API為TopologyProvider，需要了解其原理的可以參考具體的實現。

對於集群（Cluster）模式，Lettuce提供了一套獨立的API。

另外，如果Lettuce連接面向的是非單個Redis節點，連接實例提供了數據讀取節點偏好（ReadFrom）設置，可選值有：

• MASTER：只從Master節點中讀取。
• MASTER_PREFERRED：優先從Master節點中讀取。
• SLAVE_PREFERRED：優先從Slavor節點中讀取。
• SLAVE：只從Slavor節點中讀取。
• NEAREST：使用最近一次連接的Redis實例讀取。

普通主從模式#

假設現在有三個Redis服務形成樹狀主從關系如下：

• 節點一：localhost:6379，角色為Master。
• 節點二：localhost:6380，角色為Slavor，節點一的從節點。
• 節點三：localhost:6381，角色為Slavor，節點二的從節點。

首次動態節點發現主從模式的節點信息需要如下構建連接：

Copy

@Test public void testDynamicReplica() throws Exception { // 這里只需要配置一個節點的連接信息，不一定需要是主節點的信息，從節點也可以 RedisURI uri = RedisURI.builder().withHost("localhost").withPort(6379).build(); RedisClient redisClient = RedisClient.create(uri); StatefulRedisMasterSlaveConnection<String, String> connection = MasterSlave.connect(redisClient, new Utf8StringCodec(), uri); // 只從從節點讀取數據 connection.setReadFrom(ReadFrom.SLAVE); // 執行其他Redis命令 connection.close(); redisClient.shutdown(); } 

如果需要指定靜態的Redis主從節點連接屬性，那么可以這樣構建連接：

Copy

@Test public void testStaticReplica() throws Exception { List<RedisURI> uris = new ArrayList<>(); RedisURI uri1 = RedisURI.builder().withHost("localhost").withPort(6379).build(); RedisURI uri2 = RedisURI.builder().withHost("localhost").withPort(6380).build(); RedisURI uri3 = RedisURI.builder().withHost("localhost").withPort(6381).build(); uris.add(uri1); uris.add(uri2); uris.add(uri3); RedisClient redisClient = RedisClient.create(); StatefulRedisMasterSlaveConnection<String, String> connection = MasterSlave.connect(redisClient, new Utf8StringCodec(), uris); // 只從主節點讀取數據 connection.setReadFrom(ReadFrom.MASTER); // 執行其他Redis命令 connection.close(); redisClient.shutdown(); } 

哨兵模式#

由於Lettuce自身提供了哨兵的拓撲發現機制，所以只需要隨便配置一個哨兵節點的RedisURI實例即可：

Copy

@Test public void testDynamicSentinel() throws Exception { RedisURI redisUri = RedisURI.builder() .withPassword("你的密碼") .withSentinel("localhost", 26379) .withSentinelMasterId("哨兵Master的ID") .build(); RedisClient redisClient = RedisClient.create(); StatefulRedisMasterSlaveConnection<String, String> connection = MasterSlave.connect(redisClient, new Utf8StringCodec(), redisUri); // 只允許從從節點讀取數據 connection.setReadFrom(ReadFrom.SLAVE); RedisCommands<String, String> command = connection.sync(); SetArgs setArgs = SetArgs.Builder.nx().ex(5); command.set("name", "throwable", setArgs); String value = command.get("name"); log.info("Get value:{}", value); } // Get value:throwable 

集群模式#

鑒於筆者對Redis集群模式並不熟悉，Cluster模式下的API使用本身就有比較多的限制，所以這里只簡單介紹一下怎么用。先說幾個特性：

下面的API提供跨槽位（Slot）調用的功能：

• RedisAdvancedClusterCommands。
• RedisAdvancedClusterAsyncCommands。
• RedisAdvancedClusterReactiveCommands。

靜態節點選擇功能：

• masters：選擇所有主節點執行命令。
• slaves：選擇所有從節點執行命令，其實就是只讀模式。
• all nodes：命令可以在所有節點執行。

集群拓撲視圖動態更新功能：

• 手動更新，主動調用RedisClusterClient#reloadPartitions()。
• 后台定時更新。
• 自適應更新，基於連接斷開和MOVED/ASK命令重定向自動更新。

Redis集群搭建詳細過程可以參考官方文檔，假設已經搭建好集群如下（192.168.56.200是筆者的虛擬機Host）：

• 192.168.56.200:7001 => 主節點，槽位0-5460。
• 192.168.56.200:7002 => 主節點，槽位5461-10922。
• 192.168.56.200:7003 => 主節點，槽位10923-16383。
• 192.168.56.200:7004 => 7001的從節點。
• 192.168.56.200:7005 => 7002的從節點。
• 192.168.56.200:7006 => 7003的從節點。

簡單的集群連接和使用方式如下：

Copy

@Test public void testSyncCluster(){ RedisURI uri = RedisURI.builder().withHost("192.168.56.200").build(); RedisClusterClient redisClusterClient = RedisClusterClient.create(uri); StatefulRedisClusterConnection<String, String> connection = redisClusterClient.connect(); RedisAdvancedClusterCommands<String, String> commands = connection.sync(); commands.setex("name",10, "throwable"); String value = commands.get("name"); log.info("Get value:{}", value); } // Get value:throwable 

節點選擇：

Copy

@Test public void testSyncNodeSelection() { RedisURI uri = RedisURI.builder().withHost("192.168.56.200").withPort(7001).build(); RedisClusterClient redisClusterClient = RedisClusterClient.create(uri); StatefulRedisClusterConnection<String, String> connection = redisClusterClient.connect(); RedisAdvancedClusterCommands<String, String> commands = connection.sync(); // commands.all(); // 所有節點 // commands.masters(); // 主節點 // 從節點只讀 NodeSelection<String, String> replicas = commands.slaves(); NodeSelectionCommands<String, String> nodeSelectionCommands = replicas.commands(); // 這里只是演示,一般應該禁用keys *命令 Executions<List<String>> keys = nodeSelectionCommands.keys("*"); keys.forEach(key -> log.info("key: {}", key)); connection.close(); redisClusterClient.shutdown(); } 

定時更新集群拓撲視圖（每隔十分鍾更新一次，這個時間自行考量，不能太頻繁）：

Copy

@Test public void testPeriodicClusterTopology() throws Exception { RedisURI uri = RedisURI.builder().withHost("192.168.56.200").withPort(7001).build(); RedisClusterClient redisClusterClient = RedisClusterClient.create(uri); ClusterTopologyRefreshOptions options = ClusterTopologyRefreshOptions .builder() .enablePeriodicRefresh(Duration.of(10, ChronoUnit.MINUTES)) .build(); redisClusterClient.setOptions(ClusterClientOptions.builder().topologyRefreshOptions(options).build()); StatefulRedisClusterConnection<String, String> connection = redisClusterClient.connect(); RedisAdvancedClusterCommands<String, String> commands = connection.sync(); commands.setex("name", 10, "throwable"); String value = commands.get("name"); log.info("Get value:{}", value); Thread.sleep(Integer.MAX_VALUE); connection.close(); redisClusterClient.shutdown(); } 

自適應更新集群拓撲視圖：

Copy

@Test public void testAdaptiveClusterTopology() throws Exception { RedisURI uri = RedisURI.builder().withHost("192.168.56.200").withPort(7001).build(); RedisClusterClient redisClusterClient = RedisClusterClient.create(uri); ClusterTopologyRefreshOptions options = ClusterTopologyRefreshOptions.builder() .enableAdaptiveRefreshTrigger( ClusterTopologyRefreshOptions.RefreshTrigger.MOVED_REDIRECT, ClusterTopologyRefreshOptions.RefreshTrigger.PERSISTENT_RECONNECTS ) .adaptiveRefreshTriggersTimeout(Duration.of(30, ChronoUnit.SECONDS)) .build(); redisClusterClient.setOptions(ClusterClientOptions.builder().topologyRefreshOptions(options).build()); StatefulRedisClusterConnection<String, String> connection = redisClusterClient.connect(); RedisAdvancedClusterCommands<String, String> commands = connection.sync(); commands.setex("name", 10, "throwable"); String value = commands.get("name"); log.info("Get value:{}", value); Thread.sleep(Integer.MAX_VALUE); connection.close(); redisClusterClient.shutdown(); } 

動態命令和自定義命令#

自定義命令是Redis命令有限集，不過可以更細粒度指定KEY、ARGV、命令類型、編碼解碼器和返回值類型，依賴於dispatch()方法：

Copy

// 自定義實現PING方法 @Test public void testCustomPing() throws Exception { RedisURI redisUri = RedisURI.builder() .withHost("localhost") .withPort(6379) .withTimeout(Duration.of(10, ChronoUnit.SECONDS)) .build(); RedisClient redisClient = RedisClient.create(redisUri); StatefulRedisConnection<String, String> connect = redisClient.connect(); RedisCommands<String, String> sync = connect.sync(); RedisCodec<String, String> codec = StringCodec.UTF8; String result = sync.dispatch(CommandType.PING, new StatusOutput<>(codec)); log.info("PING:{}", result); connect.close(); redisClient.shutdown(); } // PING:PONG // 自定義實現Set方法 @Test public void testCustomSet() throws Exception { RedisURI redisUri = RedisURI.builder() .withHost("localhost") .withPort(6379) .withTimeout(Duration.of(10, ChronoUnit.SECONDS)) .build(); RedisClient redisClient = RedisClient.create(redisUri); StatefulRedisConnection<String, String> connect = redisClient.connect(); RedisCommands<String, String> sync = connect.sync(); RedisCodec<String, String> codec = StringCodec.UTF8; sync.dispatch(CommandType.SETEX, new StatusOutput<>(codec), new CommandArgs<>(codec).addKey("name").add(5).addValue("throwable")); String result = sync.get("name"); log.info("Get value:{}", result); connect.close(); redisClient.shutdown(); } // Get value:throwable 

動態命令是基於Redis命令有限集，並且通過注解和動態代理完成一些復雜命令組合的實現。主要注解在io.lettuce.core.dynamic.annotation包路徑下。簡單舉個例子：

Copy

public interface CustomCommand extends Commands { // SET [key] [value] @Command("SET ?0 ?1") String setKey(String key, String value); // SET [key] [value] @Command("SET :key :value") String setKeyNamed(@Param("key") String key, @Param("value") String value); // MGET [key1] [key2] @Command("MGET ?0 ?1") List<String> mGet(String key1, String key2); /** * 方法名作為命令 */ @CommandNaming(strategy = CommandNaming.Strategy.METHOD_NAME) String mSet(String key1, String value1, String key2, String value2); } @Test public void testCustomDynamicSet() throws Exception { RedisURI redisUri = RedisURI.builder() .withHost("localhost") .withPort(6379) .withTimeout(Duration.of(10, ChronoUnit.SECONDS)) .build(); RedisClient redisClient = RedisClient.create(redisUri); StatefulRedisConnection<String, String> connect = redisClient.connect(); RedisCommandFactory commandFactory = new RedisCommandFactory(connect); CustomCommand commands = commandFactory.getCommands(CustomCommand.class); commands.setKey("name", "throwable"); commands.setKeyNamed("throwable", "doge"); log.info("MGET ===> " + commands.mGet("name", "throwable")); commands.mSet("key1", "value1","key2", "value2"); log.info("MGET ===> " + commands.mGet("key1", "key2")); connect.close(); redisClient.shutdown(); } // MGET ===> [throwable, doge] // MGET ===> [value1, value2] 

高階特性#

Lettuce有很多高階使用特性，這里只列舉個人認為常用的兩點：

• 配置客戶端資源。
• 使用連接池。

更多其他特性可以自行參看官方文檔。

配置客戶端資源#

客戶端資源的設置與Lettuce的性能、並發和事件處理相關。線程池或者線程組相關配置占據客戶端資源配置的大部分（EventLoopGroups和EventExecutorGroup），這些線程池或者線程組是連接程序的基礎組件。一般情況下，客戶端資源應該在多個Redis客戶端之間共享，並且在不再使用的時候需要自行關閉。筆者認為，客戶端資源是面向Netty的。注意：除非特別熟悉或者花長時間去測試調整下面提到的參數，否則在沒有經驗的前提下憑直覺修改默認值，有可能會踩坑。

客戶端資源接口是ClientResources，實現類是DefaultClientResources。

構建DefaultClientResources實例：

Copy

// 默認 ClientResources resources = DefaultClientResources.create(); // 建造器 ClientResources resources = DefaultClientResources.builder() .ioThreadPoolSize(4) .computationThreadPoolSize(4) .build() 

使用：

Copy

ClientResources resources = DefaultClientResources.create();
// 非集群 RedisClient client = RedisClient.create(resources, uri); // 集群 RedisClusterClient clusterClient = RedisClusterClient.create(resources, uris); // ...... client.shutdown(); clusterClient.shutdown(); // 關閉資源 resources.shutdown(); 

客戶端資源基本配置：

屬性	描述	默認值
ioThreadPoolSize	I/O線程數	Runtime.getRuntime().availableProcessors()
computationThreadPoolSize	任務線程數	Runtime.getRuntime().availableProcessors()

客戶端資源高級配置：

屬性	描述	默認值
eventLoopGroupProvider	EventLoopGroup提供商	-
eventExecutorGroupProvider	EventExecutorGroup提供商	-
eventBus	事件總線	DefaultEventBus
commandLatencyCollectorOptions	命令延時收集器配置	DefaultCommandLatencyCollectorOptions
commandLatencyCollector	命令延時收集器	DefaultCommandLatencyCollector
commandLatencyPublisherOptions	命令延時發布器配置	DefaultEventPublisherOptions
dnsResolver	DNS處理器	JDK或者Netty提供
reconnectDelay	重連延時配置	Delay.exponential()
nettyCustomizer	Netty自定義配置器	-
tracing	軌跡記錄器	-

非集群客戶端RedisClient的屬性配置：

Redis非集群客戶端RedisClient本身提供了配置屬性方法：

Copy

RedisClient client = RedisClient.create(uri);
client.setOptions(ClientOptions.builder()
                       .autoReconnect(false) .pingBeforeActivateConnection(true) .build()); 

非集群客戶端的配置屬性列表：

屬性	描述	默認值
pingBeforeActivateConnection	連接激活之前是否執行PING命令	false
autoReconnect	是否自動重連	true
cancelCommandsOnReconnectFailure	重連失敗是否拒絕命令執行	false
suspendReconnectOnProtocolFailure	底層協議失敗是否掛起重連操作	false
requestQueueSize	請求隊列容量	2147483647(Integer#MAX_VALUE)
disconnectedBehavior	失去連接時候的行為	DEFAULT
sslOptions	SSL配置	-
socketOptions	Socket配置	10 seconds Connection-Timeout, no keep-alive, no TCP noDelay
timeoutOptions	超時配置	-
publishOnScheduler	發布反應式信號數據的調度器	使用I/O線程

集群客戶端屬性配置：

Redis集群客戶端RedisClusterClient本身提供了配置屬性方法：

Copy

RedisClusterClient client = RedisClusterClient.create(uri);
ClusterTopologyRefreshOptions topologyRefreshOptions = ClusterTopologyRefreshOptions.builder()
                .enablePeriodicRefresh(refreshPeriod(10, TimeUnit.MINUTES)) .enableAllAdaptiveRefreshTriggers() .build(); client.setOptions(ClusterClientOptions.builder() .topologyRefreshOptions(topologyRefreshOptions) .build()); 

集群客戶端的配置屬性列表：

屬性	描述	默認值
enablePeriodicRefresh	是否允許周期性更新集群拓撲視圖	false
refreshPeriod	更新集群拓撲視圖周期	60秒
enableAdaptiveRefreshTrigger	設置自適應更新集群拓撲視圖觸發器RefreshTrigger	-
adaptiveRefreshTriggersTimeout	自適應更新集群拓撲視圖觸發器超時設置	30秒
refreshTriggersReconnectAttempts	自適應更新集群拓撲視圖觸發重連次數	5
dynamicRefreshSources	是否允許動態刷新拓撲資源	true
closeStaleConnections	是否允許關閉陳舊的連接	true
maxRedirects	集群重定向次數上限	5
validateClusterNodeMembership	是否校驗集群節點的成員關系	true

使用連接池#

引入連接池依賴commons-pool2：

Copy

<dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-pool2</artifactId> <version>2.7.0</version> </dependency 

基本使用如下：

Copy

@Test public void testUseConnectionPool() throws Exception { RedisURI redisUri = RedisURI.builder() .withHost("localhost") .withPort(6379) .withTimeout(Duration.of(10, ChronoUnit.SECONDS)) .build(); RedisClient redisClient = RedisClient.create(redisUri); GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig(); GenericObjectPool<StatefulRedisConnection<String, String>> pool = ConnectionPoolSupport.createGenericObjectPool(redisClient::connect, poolConfig); try (StatefulRedisConnection<String, String> connection = pool.borrowObject()) { RedisCommands<String, String> command = connection.sync(); SetArgs setArgs = SetArgs.Builder.nx().ex(5); command.set("name", "throwable", setArgs); String n = command.get("name"); log.info("Get value:{}", n); } pool.close(); redisClient.shutdown(); } 

其中，同步連接的池化支持需要用ConnectionPoolSupport，異步連接的池化支持需要用AsyncConnectionPoolSupport（Lettuce5.1之后才支持）。

幾個常見的漸進式刪除例子#

漸進式刪除Hash中的域-屬性：

Copy

@Test public void testDelBigHashKey() throws Exception { // SCAN參數 ScanArgs scanArgs = ScanArgs.Builder.limit(2); // TEMP游標 ScanCursor cursor = ScanCursor.INITIAL; // 目標KEY String key = "BIG_HASH_KEY"; prepareHashTestData(key); log.info("開始漸進式刪除Hash的元素..."); int counter = 0; do { MapScanCursor<String, String> result = COMMAND.hscan(key, cursor, scanArgs); // 重置TEMP游標 cursor = ScanCursor.of(result.getCursor()); cursor.setFinished(result.isFinished()); Collection<String> fields = result.getMap().values(); if (!fields.isEmpty()) { COMMAND.hdel(key, fields.toArray(new String[0])); } counter++; } while (!(ScanCursor.FINISHED.getCursor().equals(cursor.getCursor()) && ScanCursor.FINISHED.isFinished() == cursor.isFinished())); log.info("漸進式刪除Hash的元素完畢,迭代次數:{} ...", counter); } private void prepareHashTestData(String key) throws Exception { COMMAND.hset(key, "1", "1"); COMMAND.hset(key, "2", "2"); COMMAND.hset(key, "3", "3"); COMMAND.hset(key, "4", "4"); COMMAND.hset(key, "5", "5"); } 

漸進式刪除集合中的元素：

Copy

@Test public void testDelBigSetKey() throws Exception { String key = "BIG_SET_KEY"; prepareSetTestData(key); // SCAN參數 ScanArgs scanArgs = ScanArgs.Builder.limit(2); // TEMP游標 ScanCursor cursor = ScanCursor.INITIAL; log.info("開始漸進式刪除Set的元素..."); int counter = 0; do { ValueScanCursor<String> result = COMMAND.sscan(key, cursor, scanArgs); // 重置TEMP游標 cursor = ScanCursor.of(result.getCursor()); cursor.setFinished(result.isFinished()); List<String> values = result.getValues(); if (!values.isEmpty()) { COMMAND.srem(key, values.toArray(new String[0])); } counter++; } while (!(ScanCursor.FINISHED.getCursor().equals(cursor.getCursor()) && ScanCursor.FINISHED.isFinished() == cursor.isFinished())); log.info("漸進式刪除Set的元素完畢,迭代次數:{} ...", counter); } private void prepareSetTestData(String key) throws Exception { COMMAND.sadd(key, "1", "2", "3", "4", "5"); } 

漸進式刪除有序集合中的元素：

Copy

@Test public void testDelBigZSetKey() throws Exception { // SCAN參數 ScanArgs scanArgs = ScanArgs.Builder.limit(2); // TEMP游標 ScanCursor cursor = ScanCursor.INITIAL; // 目標KEY String key = "BIG_ZSET_KEY"; prepareZSetTestData(key); log.info("開始漸進式刪除ZSet的元素..."); int counter = 0; do { ScoredValueScanCursor<String> result = COMMAND.zscan(key, cursor, scanArgs); // 重置TEMP游標 cursor = ScanCursor.of(result.getCursor()); cursor.setFinished(result.isFinished()); List<ScoredValue<String>> scoredValues = result.getValues(); if (!scoredValues.isEmpty()) { COMMAND.zrem(key, scoredValues.stream().map(ScoredValue<String>::getValue).toArray(String[]::new)); } counter++; } while (!(ScanCursor.FINISHED.getCursor().equals(cursor.getCursor()) && ScanCursor.FINISHED.isFinished() == cursor.isFinished())); log.info("漸進式刪除ZSet的元素完畢,迭代次數:{} ...", counter); } private void prepareZSetTestData(String key) throws Exception { COMMAND.zadd(key, 0, "1"); COMMAND.zadd(key, 0, "2"); COMMAND.zadd(key, 0, "3"); COMMAND.zadd(key, 0, "4"); COMMAND.zadd(key, 0, "5"); } 

在SpringBoot中使用Lettuce#

個人認為，spring-data-redis中的API封裝並不是很優秀，用起來比較重，不夠靈活，這里結合前面的例子和代碼，在SpringBoot腳手架項目中配置和整合Lettuce。先引入依賴：

Copy

<dependencyManagement> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-dependencies</artifactId> <version>2.1.8.RELEASE</version> <type>pom</type> <scope>import</scope> </dependency> </dependencies> </dependencyManagement> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>io.lettuce</groupId> <artifactId>lettuce-core</artifactId> <version>5.1.8.RELEASE</version> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <version>1.18.10</version> <scope>provided</scope> </dependency> </dependencies> 

一般情況下，每個應用應該使用單個Redis客戶端實例和單個連接實例，這里設計一個腳手架，適配單機、普通主從、哨兵和集群四種使用場景。對於客戶端資源，采用默認的實現即可。對於Redis的連接屬性，比較主要的有Host、Port和Password，其他可以暫時忽略。基於約定大於配置的原則，先定制一系列屬性配置類（其實有些配置是可以完全共用，但是考慮到要清晰描述類之間的關系，這里拆分多個配置屬性類和多個配置方法）：

Copy

@Data @ConfigurationProperties(prefix = "lettuce") public class LettuceProperties { private LettuceSingleProperties single; private LettuceReplicaProperties replica; private LettuceSentinelProperties sentinel; private LettuceClusterProperties cluster; } @Data public class LettuceSingleProperties { private String host; private Integer port; private String password; } @EqualsAndHashCode(callSuper = true) @Data public class LettuceReplicaProperties extends LettuceSingleProperties { } @EqualsAndHashCode(callSuper = true) @Data public class LettuceSentinelProperties extends LettuceSingleProperties { private String masterId; } @EqualsAndHashCode(callSuper = true) @Data public class LettuceClusterProperties extends LettuceSingleProperties { } 

配置類如下，主要使用@ConditionalOnProperty做隔離，一般情況下，很少有人會在一個應用使用一種以上的Redis連接場景：

Copy

@RequiredArgsConstructor @Configuration @ConditionalOnClass(name = "io.lettuce.core.RedisURI") @EnableConfigurationProperties(value = LettuceProperties.class) public class LettuceAutoConfiguration { private final LettuceProperties lettuceProperties; @Bean(destroyMethod = "shutdown") public ClientResources clientResources() { return DefaultClientResources.create(); } @Bean @ConditionalOnProperty(name = "lettuce.single.host") public RedisURI singleRedisUri() { LettuceSingleProperties singleProperties = lettuceProperties.getSingle(); return RedisURI.builder() .withHost(singleProperties.getHost()) .withPort(singleProperties.getPort()) .withPassword(singleProperties.getPassword()) .build(); } @Bean(destroyMethod = "shutdown") @ConditionalOnProperty(name = "lettuce.single.host") public RedisClient singleRedisClient(ClientResources clientResources, @Qualifier("singleRedisUri") RedisURI redisUri) { return RedisClient.create(clientResources, redisUri); } @Bean(destroyMethod = "close") @ConditionalOnProperty(name = "lettuce.single.host") public StatefulRedisConnection<String, String> singleRedisConnection(@Qualifier("singleRedisClient") RedisClient singleRedisClient) { return singleRedisClient.connect(); } @Bean @ConditionalOnProperty(name = "lettuce.replica.host") public RedisURI replicaRedisUri() { LettuceReplicaProperties replicaProperties = lettuceProperties.getReplica(); return RedisURI.builder() .withHost(replicaProperties.getHost()) .withPort(replicaProperties.getPort()) .withPassword(replicaProperties.getPassword()) .build(); } @Bean(destroyMethod = "shutdown") @ConditionalOnProperty(name = "lettuce.replica.host") public RedisClient replicaRedisClient(ClientResources clientResources, @Qualifier("replicaRedisUri") RedisURI redisUri) { return RedisClient.create(clientResources, redisUri); } @Bean(destroyMethod = "close") @ConditionalOnProperty(name = "lettuce.replica.host") public StatefulRedisMasterSlaveConnection<String, String> replicaRedisConnection(@Qualifier("replicaRedisClient") RedisClient replicaRedisClient, @Qualifier("replicaRedisUri") RedisURI redisUri) { return MasterSlave.connect(replicaRedisClient, new Utf8StringCodec(), redisUri); } @Bean @ConditionalOnProperty(name = "lettuce.sentinel.host") public RedisURI sentinelRedisUri() { LettuceSentinelProperties sentinelProperties = lettuceProperties.getSentinel(); return RedisURI.builder() .withPassword(sentinelProperties.getPassword()) .withSentinel(sentinelProperties.getHost(), sentinelProperties.getPort()) .withSentinelMasterId(sentinelProperties.getMasterId()) .build(); } @Bean(destroyMethod = "shutdown") @ConditionalOnProperty(name = "lettuce.sentinel.host") public RedisClient sentinelRedisClient(ClientResources clientResources, @Qualifier("sentinelRedisUri") RedisURI redisUri) { return RedisClient.create(clientResources, redisUri); } @Bean(destroyMethod = "close") @ConditionalOnProperty(name = "lettuce.sentinel.host") public StatefulRedisMasterSlaveConnection<String, String> sentinelRedisConnection(@Qualifier("sentinelRedisClient") RedisClient sentinelRedisClient, @Qualifier("sentinelRedisUri") RedisURI redisUri) { return MasterSlave.connect(sentinelRedisClient, new Utf8StringCodec(), redisUri); } @Bean @ConditionalOnProperty(name = "lettuce.cluster.host") public RedisURI clusterRedisUri() { LettuceClusterProperties clusterProperties = lettuceProperties.getCluster(); return RedisURI.builder() .withHost(clusterProperties.getHost()) .withPort(clusterProperties.getPort()) .withPassword(clusterProperties.getPassword()) .build(); } @Bean(destroyMethod = "shutdown") @ConditionalOnProperty(name = "lettuce.cluster.host") public RedisClusterClient redisClusterClient(ClientResources clientResources, @Qualifier("clusterRedisUri") RedisURI redisUri) { return RedisClusterClient.create(clientResources, redisUri); } @Bean(destroyMethod = "close") @ConditionalOnProperty(name = "lettuce.cluster") public StatefulRedisClusterConnection<String, String> clusterConnection(RedisClusterClient clusterClient) { return clusterClient.connect(); } } 

最后為了讓IDE識別我們的配置，可以添加IDE親緣性，/META-INF文件夾下新增一個文件spring-configuration-metadata.json，內容如下：

Copy

{
  "properties": [ { "name": "lettuce.single", "type": "club.throwable.spring.lettuce.LettuceSingleProperties", "description": "單機配置", "sourceType": "club.throwable.spring.lettuce.LettuceProperties" }, { "name": "lettuce.replica", "type": "club.throwable.spring.lettuce.LettuceReplicaProperties", "description": "主從配置", "sourceType": "club.throwable.spring.lettuce.LettuceProperties" }, { "name": "lettuce.sentinel", "type": "club.throwable.spring.lettuce.LettuceSentinelProperties", "description": "哨兵配置", "sourceType": "club.throwable.spring.lettuce.LettuceProperties" }, { "name": "lettuce.single", "type": "club.throwable.spring.lettuce.LettuceClusterProperties", "description": "集群配置", "sourceType": "club.throwable.spring.lettuce.LettuceProperties" } ] } 

如果想IDE親緣性做得更好，可以添加/META-INF/additional-spring-configuration-metadata.json進行更多細節定義。簡單使用如下：

Copy

@Slf4j @Component public class RedisCommandLineRunner implements CommandLineRunner { @Autowired @Qualifier("singleRedisConnection") private StatefulRedisConnection<String, String> connection; @Override public void run(String... args) throws Exception { RedisCommands<String, String> redisCommands = connection.sync(); redisCommands.setex("name", 5, "throwable"); log.info("Get value:{}", redisCommands.get("name")); } } // Get value:throwable 

小結#

本文算是基於Lettuce的官方文檔，對它的使用進行全方位的分析，包括主要功能、配置都做了一些示例，限於篇幅部分特性和配置細節沒有分析。Lettuce已經被spring-data-redis接納作為官方的Redis客戶端驅動，所以值得信賴，它的一些API設計確實比較合理，擴展性高的同時靈活性也高。個人建議，基於Lettuce包自行添加配置到SpringBoot應用用起來會得心應手，畢竟RedisTemplate實在太笨重，而且還屏蔽了Lettuce一些高級特性和靈活的API。

參考資料：

• Lettuce Reference Guide

鏈接#

• Github Page：http://www.throwable.club/2019/09/28/redis-client-driver-lettuce-usage
• Coding Page：http://throwable.coding.me/2019/09/28/redis-client-driver-lettuce-usage