Redis Lettuce連接redis集群實現過程詳細講解
前言
Lettuce連接redis集群使用的都是集群專用類,像RedisClusterClient、StatefulRedisClusterConnection、RedisAdvancedClusterCommands、StatefulRedisClusterPubSubConnection等等;
Lettuce對rediscluster的支持:
- 支持所有Cluster命令;
- 基於鍵哈希槽的路由節點;
- 對集群命令高級抽象;
- 在多個集群節點上執行命令;
- 處理MOVED和ASK重定向;
- 通過槽位和ip端口直接連接集群節點;
- SSL和身份驗證;
- 定期和自適應集群拓撲更新;
- 發佈訂閱;
啟動時隻需至少一個可以連接的集群節點就可以,能夠自動拓撲出集群全部節點;也可以使用ReadFrom設置讀取數據來源,跟主從模式一樣;
雖然redis本身的多鍵命令要求key必須都在同一個槽位,但Lettuce對一部分命令多瞭優化,可以對多鍵命令進行跨槽位執行,通過將對不同槽位鍵的操作命令分解為多條命令,單個命令以fork/join方式並發運行,最後將結果合並返回;
可以跨槽位的命令有
- DEL:刪除鍵,返回刪除數量;
- EXISTS:統計跨槽位的存在的鍵的數量;
- MGET:獲取所有給定鍵的值,順序按照鍵的順序返回;
- MSET:批量保存鍵值對,總是返回OK;
- TOUCH:改變給定鍵的最後訪問時間,返回改變的鍵的數量;
- UNLINK:刪除鍵並在另一個不同的線程中回收內存,返回刪除數量;
提供跨槽位命令的api:RedisAdvancedClusterCommands、RedisAdvancedClusterAsyncCommands、RedisAdvancedClusterReactiveCommands;
可以在多個集群節點上執行的命令有
- CLIENT SETNAME:在所有已知的集群節點上設置客戶端的名稱,總是返回OK;
- KEYS:返回所有master上存儲的key;
- DBSIZE:返回所有master上存儲的key的數量;
- FLUSHALL:清空master上的所有數據,總是返回OK;
- FLUSHDB:清空master上的所有數據,總是返回OK;
- RANDOMKEY:從隨機master上返回隨機的key;
- SCAN:根據ReadFrom設置掃描整個集群的鍵空間;
- SCRIPT FLUSH:從所有的集群節點腳本緩存中刪除所有腳本;
- SCRIPT LOAD:在所有的集群節點上加載lua腳本;
- SCRIPT KILL:在所有集群節點上殺死腳本;(即使腳本沒有運行調用也不會失敗)
- SHUTDOWN:將數據集同步保存到磁盤,然後關閉集群所有節點;
關於發佈訂閱
普通用戶空間的發佈訂閱,redis集群會發送到每個節點,發佈者和訂閱者不需要在同一個節點,普通訂閱發佈消息可以在集群拓撲改變時重新連接。對於鍵空間事件,隻會發到自己的節點,不會擴散到其他節點,要訂閱鍵空間事件可以去適當的多個節點上訂閱,或者使用RedisClusterClient消息傳播和NodeSelectionAPI獲得一個托管連接集合;
註意:由於主從同步,鍵會被復制到多個從節點上,特別是鍵過期事件,會在主從節點上都產生過期事件,如果訂閱從節點,可能會收到多條相同的過期事件;訂閱是通過NodeSelectionAPI或者單個節點調用subscribe(…)發出的,訂閱對於新增的節點無效;
測試Demo:(redis版本7.0.2,Lettuce版本6.1.8)
集群節點:虛擬機 192.168.1.31,端口 9001-9006,集群節點已設置notify-keyspace-events AK;
package testlettuce; import java.time.Duration; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import io.lettuce.core.ClientOptions.DisconnectedBehavior; import io.lettuce.core.KeyScanCursor; import io.lettuce.core.KeyValue; import io.lettuce.core.ReadFrom; import io.lettuce.core.RedisURI; import io.lettuce.core.ScanCursor; import io.lettuce.core.SocketOptions; import io.lettuce.core.SslOptions; import io.lettuce.core.TimeoutOptions; import io.lettuce.core.cluster.ClusterClientOptions; import io.lettuce.core.cluster.ClusterTopologyRefreshOptions; import io.lettuce.core.cluster.RedisClusterClient; import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; import io.lettuce.core.cluster.api.sync.Executions; import io.lettuce.core.cluster.api.sync.NodeSelection; import io.lettuce.core.cluster.api.sync.NodeSelectionCommands; import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands; import io.lettuce.core.cluster.pubsub.StatefulRedisClusterPubSubConnection; import io.lettuce.core.cluster.pubsub.api.async.NodeSelectionPubSubAsyncCommands; import io.lettuce.core.cluster.pubsub.api.async.PubSubAsyncNodeSelection; import io.lettuce.core.cluster.pubsub.api.reactive.RedisClusterPubSubReactiveCommands; import io.lettuce.core.protocol.DecodeBufferPolicies; import io.lettuce.core.protocol.ProtocolVersion; import io.lettuce.core.pubsub.RedisPubSubListener; import io.lettuce.core.pubsub.api.async.RedisPubSubAsyncCommands; public class TestLettuceCluster { /** * @param args */ public static void main(String[] args) { List<RedisURI> nodeList = new ArrayList<>(); nodeList.add(RedisURI.builder().withHost("192.168.1.31").withPort(9001).withAuthentication("default", "123456").build()); nodeList.add(RedisURI.builder().withHost("192.168.1.31").withPort(9002).withAuthentication("default", "123456").build()); nodeList.add(RedisURI.builder().withHost("192.168.1.31").withPort(9003).withAuthentication("default", "123456").build()); nodeList.add(RedisURI.builder().withHost("192.168.1.31").withPort(9004).withAuthentication("default", "123456").build()); nodeList.add(RedisURI.builder().withHost("192.168.1.31").withPort(9005).withAuthentication("default", "123456").build()); nodeList.add(RedisURI.builder().withHost("192.168.1.31").withPort(9006).withAuthentication("default", "123456").build()); RedisClusterClient clusterClient = RedisClusterClient.create(nodeList); ClusterTopologyRefreshOptions clusterTopologyRefreshOptions = ClusterTopologyRefreshOptions.builder() .adaptiveRefreshTriggersTimeout(Duration.ofSeconds(5L))//設置自適應拓撲刷新超時,每次超時刷新一次,默認30s; .closeStaleConnections(false)//刷新拓撲時是否關閉失效連接,默認true,isPeriodicRefreshEnabled()為true時生效; .dynamicRefreshSources(true)//從拓撲中發現新節點,並將新節點也作為拓撲的源節點,動態刷新可以發現全部節點並計算每個客戶端的數量,設置false則隻有初始節點為源和計算客戶端數量; .enableAllAdaptiveRefreshTriggers()//啟用全部觸發器自適應刷新拓撲,默認關閉; .enablePeriodicRefresh(Duration.ofSeconds(5L))//開啟定時拓撲刷新並設置周期; .refreshTriggersReconnectAttempts(3)//長連接重新連接嘗試n次才拓撲刷新 .build(); ClusterClientOptions clusterClientOptions = ClusterClientOptions.builder() .autoReconnect(true)//在連接丟失時開啟或關閉自動重連,默認true; .cancelCommandsOnReconnectFailure(true)//允許在重連失敗取消排隊命令,默認false; .decodeBufferPolicy(DecodeBufferPolicies.always())//設置丟棄解碼緩沖區的策略,以回收內存;always:解碼後丟棄,最大內存效率;alwaysSome:解碼後丟棄一部分;ratio(n)基於比率丟棄,n/(1+n),通常用1-10對應50%-90%; .disconnectedBehavior(DisconnectedBehavior.DEFAULT)//設置連接斷開時命令的調用行為,默認啟用重連;DEFAULT:啟用時重連中接收命令,禁用時重連中拒絕命令;ACCEPT_COMMANDS:重連中接收命令;REJECT_COMMANDS:重連中拒絕命令; // .maxRedirects(5)//當鍵從一個節點遷移到另一個節點,集群重定向次數,默認5; // .nodeFilter(nodeFilter)//設置節點過濾器 // .pingBeforeActivateConnection(true)//激活連接前設置PING,默認true; // .protocolVersion(ProtocolVersion.RESP3)//設置協議版本,默認RESP3; // .publishOnScheduler(false)//使用專用的調度器發出響應信號,默認false,啟用時數據信號將使用服務的多線程發出; // .requestQueueSize(requestQueueSize)//設置每個連接請求隊列大小; // .scriptCharset(scriptCharset)//設置Lua腳本編碼為byte[]的字符集,默認StandardCharsets.UTF_8; // .socketOptions(SocketOptions.builder().connectTimeout(Duration.ofSeconds(10)).keepAlive(true).tcpNoDelay(true).build())//設置低級套接字的屬性 // .sslOptions(SslOptions.builder().build())//設置ssl屬性 // .suspendReconnectOnProtocolFailure(false)//當重新連接遇到協議失敗時暫停重新連接(SSL驗證,連接失敗前PING),默認值為false; // .timeoutOptions(TimeoutOptions.enabled(Duration.ofSeconds(10)))//設置超時來取消和終止命令; .topologyRefreshOptions(clusterTopologyRefreshOptions)//設置拓撲更新設置 .validateClusterNodeMembership(true)//在允許連接到集群節點之前,驗證集群節點成員關系,默認值為true; .build(); clusterClient.setDefaultTimeout(Duration.ofSeconds(5L)); clusterClient.setOptions(clusterClientOptions); StatefulRedisClusterConnection<String, String> clusterConn = clusterClient.connect(); clusterConn.setReadFrom(ReadFrom.ANY);//設置從哪些節點讀取數據; RedisAdvancedClusterCommands<String, String> clusterCmd = clusterConn.sync(); clusterCmd.set("a", "A"); clusterCmd.set("b", "B"); clusterCmd.set("c", "C"); clusterCmd.set("d", "D"); System.out.println("get a=" + clusterCmd.get("a")); System.out.println("get b=" + clusterCmd.get("b")); System.out.println("get c=" + clusterCmd.get("c")); System.out.println("get d=" + clusterCmd.get("d")); //跨槽位命令 Map<String, String> kvmap = new HashMap<>(); kvmap.put("a", "AA"); kvmap.put("b", "BB"); kvmap.put("c", "CC"); kvmap.put("d", "DD"); clusterCmd.mset(kvmap);//Lettuce做瞭優化,支持一些命令的跨槽位命令; System.out.println("Lettuce mget:" + clusterCmd.mget("a", "b", "c", "d")); //選定部分節點操作 NodeSelection<String, String> replicas = clusterCmd.replicas(); NodeSelectionCommands<String, String> replicaseCmd = replicas.commands(); Executions<KeyScanCursor<String>> executions = replicaseCmd.scan(ScanCursor.INITIAL); executions.forEach(s -> {System.out.println(s.getKeys());}); //訂閱發佈消息 StatefulRedisClusterPubSubConnection<String, String> pubSubConn = clusterClient.connectPubSub(); pubSubConn.addListener(new RedisPubSubListener<String, String>() { @Override public void message(String channel, String message) { System.out.println("[message]ch:" + channel + ",msg:" + message); } @Override public void message(String pattern, String channel, String message) { } @Override public void subscribed(String channel, long count) { System.out.println("[subscribed]ch:" + channel); } @Override public void psubscribed(String pattern, long count) { } @Override public void unsubscribed(String channel, long count) { } @Override public void punsubscribed(String pattern, long count) { } }); pubSubConn.sync().subscribe("TEST_Ch");//(回調內部使用阻塞調用或者lettuce同步api調用,需使用異步訂閱) clusterCmd.publish("TEST_Ch", "MSGMSGMSG"); //響應式訂閱,可以監聽ChannelMessage和PatternMessage,使用鏈式過濾處理計算等操作 RedisClusterPubSubReactiveCommands<String, String> pubsubReactive = pubSubConn.reactive(); pubsubReactive.subscribe("TEST_Ch2").subscribe(); pubsubReactive.observeChannels() .filter(chmsg -> {return chmsg.getMessage().contains("tom");}) .doOnNext(chmsg -> {System.out.println("<tom>" + chmsg.getChannel() + ">>" + chmsg.getMessage());}) .subscribe(); clusterCmd.publish("TEST_Ch2", "send to jerry"); clusterCmd.publish("TEST_Ch", "tom MSG"); clusterCmd.publish("TEST_Ch2", "this is tom"); //keySpaceEvent事件 StatefulRedisClusterPubSubConnection<String, String> clusterPubSubConn = clusterClient.connectPubSub(); clusterPubSubConn.setNodeMessagePropagation(true);//啟用禁用節點消息傳播到該listener,例如隻能在本節點通知的鍵事件通知; RedisPubSubListener<String, String> listener = new RedisPubSubListener<String, String>() { @Override public void unsubscribed(String channel, long count) { System.out.println("unsubscribed_ch:" + channel); } @Override public void subscribed(String channel, long count) { System.out.println("subscribed_ch:" + channel); } @Override public void punsubscribed(String pattern, long count) { System.out.println("punsubscribed_pattern:" + pattern); } @Override public void psubscribed(String pattern, long count) { System.out.println("psubscribed_pattern:" + pattern); } @Override public void message(String pattern, String channel, String message) { System.out.println("message_pattern:" + pattern + " ch:" + channel + " msg:" + message); } @Override public void message(String channel, String message) { System.out.println("message_ch:" + channel + " msg:" + message); } }; clusterPubSubConn.addListener(listener); PubSubAsyncNodeSelection<String, String> allPubSubAsyncNodeSelection = clusterPubSubConn.async().all(); NodeSelectionPubSubAsyncCommands<String, String> pubsubAsyncCmd = allPubSubAsyncNodeSelection.commands(); clusterCmd.setex("a", 1, "A"); pubsubAsyncCmd.psubscribe("__keyspace@0__:*"); try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("end"); } }
運行結果:
另外,還有一個cluster專用的Listener:RedisClusterPubSubListener,可以從listener裡獲得發佈消息的節點信息:
RedisClusterPubSubListener<String, String> clusterListener = new RedisClusterPubSubListener<String, String>() { @Override public void message(RedisClusterNode node, String channel, String message) { } @Override public void message(RedisClusterNode node, String pattern, String channel, String message) { } @Override public void subscribed(RedisClusterNode node, String channel, long count) { } @Override public void psubscribed(RedisClusterNode node, String pattern, long count) { } @Override public void unsubscribed(RedisClusterNode node, String channel, long count) { } @Override public void punsubscribed(RedisClusterNode node, String pattern, long count) { } };
到此這篇關於Redis Lettuce連接redis集群實現過程詳細講解的文章就介紹到這瞭,更多相關Redis Lettuce連接redis集群內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- Java RabbitMQ的持久化和發佈確認詳解
- Java創建型模式之建造者模式詳解
- RabbitMQ冪等性與優先級及惰性詳細全面講解
- redis實現隊列的阻塞、延時、發佈和訂閱
- 使用Lombok @Builder註解導致默認值無效的問題