Redis高级客户端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
- Gradle
連接Redis
單機、哨兵、集群模式下連接Redis需要一個統一的標準去表示連接的細節信息,在Lettuce中這個統一的標準是RedisURI。可以通過三種方式構造一個RedisURI實例:
- 定制的字符串URI語法:
- 使用建造器(RedisURI.Builder):
- 直接通過構造函數實例化:
定制的連接URI語法
- 單機(前綴為redis://)
- 單機并且使用SSL(前綴為rediss://) <== 注意后面多了個s
- 單機Unix Domain Sockets模式(前綴為redis-socket://)
- 哨兵(前綴為redis-sentinel://)
超時時間單位:
- 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系列接口打交道。
一個基本使用例子如下:
@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連接備用:
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實例獲取,見其接口定義:
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實例:
private static RedisCommands<String, String> COMMAND;@BeforeClass public static void beforeClass() {COMMAND = CONNECTION.sync(); }基本使用:
@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實例:
private static RedisAsyncCommands<String, String> ASYNC_COMMAND;@BeforeClass public static void beforeClass() {ASYNC_COMMAND = CONNECTION.async(); }基本使用:
@Test public void testAsyncPing() throws Exception {RedisFuture<String> redisFuture = ASYNC_COMMAND.ping();log.info("Ping result:{}", redisFuture.get()); } // Ping result:PONGRedisAsyncCommands所有方法執行返回結果都是RedisFuture實例,而RedisFuture接口的定義如下:
public interface RedisFuture<V> extends CompletionStage<V>, Future<V> {String getError();boolean await(long timeout, TimeUnit unit) throws InterruptedException; }也就是,RedisFuture可以無縫使用Future或者JDK1.8中引入的CompletableFuture提供的方法。舉個例子:
@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實例:
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。舉個例子:
@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舉個更加復雜的例子,包含了事務、函數轉換等:
@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():
- 非集群模式:
- 集群模式:
這里用單機同步命令的模式舉一個Redis鍵空間通知(Redis Keyspace Notifications)的例子:
@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<>() {@Overridepublic void psubscribed(String pattern, long count) {log.info("pattern:{},count:{}", pattern, count);}@Overridepublic 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系列接口中有對應的方法。舉個例子:
// 同步模式 @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:OKRedis的Pipeline也就是管道機制可以理解為把多個命令打包在一次請求發送到Redis服務端,然后Redis服務端把所有的響應結果打包好一次性返回,從而節省不必要的網絡資源(最主要是減少網絡請求次數)。Redis對于Pipeline機制如何實現并沒有明確的規定,也沒有提供特殊的命令支持Pipeline機制。Jedis中底層采用BIO(阻塞IO)通訊,所以它的做法是客戶端緩存將要發送的命令,最后需要觸發然后同步發送一個巨大的命令列表包,再接收和解析一個巨大的響應列表包。Pipeline在Lettuce中對使用者是透明的,由于底層的通訊框架是Netty,所以網絡通訊層面的優化Lettuce不需要過多干預,換言之可以這樣理解:Netty幫Lettuce從底層實現了Redis的Pipeline機制。但是,Lettuce的異步API也提供了手動Flush的方法:
@Test public void testAsyncManualFlush() {// 取消自動flushASYNC_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執行耗時比較低:
@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命令的同步接口如下:
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()方法。舉個簡單的例子:
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,節點二的從節點。
首次動態節點發現主從模式的節點信息需要如下構建連接:
@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主從節點連接屬性,那么可以這樣構建連接:
@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實例即可:
@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的從節點。
簡單的集群連接和使用方式如下:
@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節點選擇:
@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(); }定時更新集群拓撲視圖(每隔十分鐘更新一次,這個時間自行考量,不能太頻繁):
@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(); }自適應更新集群拓撲視圖:
@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()方法:
// 自定義實現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包路徑下。簡單舉個例子:
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實例:
// 默認 ClientResources resources = DefaultClientResources.create();// 建造器 ClientResources resources = DefaultClientResources.builder().ioThreadPoolSize(4).computationThreadPoolSize(4).build()使用:
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本身提供了配置屬性方法:
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本身提供了配置屬性方法:
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:
<dependency><groupId>org.apache.commons</groupId><artifactId>commons-pool2</artifactId><version>2.7.0</version> </dependency基本使用如下:
@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中的域-屬性:
@Test public void testDelBigHashKey() throws Exception {// SCAN參數ScanArgs scanArgs = ScanArgs.Builder.limit(2);// TEMP游標ScanCursor cursor = ScanCursor.INITIAL;// 目標KEYString 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"); }漸進式刪除集合中的元素:
@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"); }漸進式刪除有序集合中的元素:
@Test public void testDelBigZSetKey() throws Exception {// SCAN參數ScanArgs scanArgs = ScanArgs.Builder.limit(2);// TEMP游標ScanCursor cursor = ScanCursor.INITIAL;// 目標KEYString 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。先引入依賴:
<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,其他可以暫時忽略。基于約定大于配置的原則,先定制一系列屬性配置類(其實有些配置是可以完全共用,但是考慮到要清晰描述類之間的關系,這里拆分多個配置屬性類和多個配置方法):
@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連接場景:
@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,內容如下:
{"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進行更多細節定義。簡單使用如下:
@Slf4j @Component public class RedisCommandLineRunner implements CommandLineRunner {@Autowired@Qualifier("singleRedisConnection")private StatefulRedisConnection<String, String> connection;@Overridepublic 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
(本文完 c-14-d e-a-20190928 最近事太多...)
轉載于:https://www.cnblogs.com/throwable/p/11601538.html
總結
以上是生活随笔為你收集整理的Redis高级客户端Lettuce详解的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: Python 常用Web框架的比较
- 下一篇: 一个低级错误引发Netty编码解码中文异