redis clients lettuce lettuce_cluster_failover

Lettuce Cluster Failover and Load Balancing

Redis Developer Course	Redis Technical Support	Redis Enterprise Server

Cluster Failover and Load Balancing

Redis Cluster에 접속하는 방법을 설명한다.
입력/조회 명령 실행 중 마스터 서버가 다운되었을 때 처리 방법에 대선 설정
조회 명령을 복제본에서 실행해서 부하 분산(load balancing)하는 방법을 설명한다.
Lettuce-6.1.7과 Redis-7.0.3에서 실행했다.

• Redis Cluster에 접속하는 방법을 설명한다.
• 입력, 조회 중 서버가 다운되었을 경우 장애 조치(failover) 테스트
• 조회 명령을 복제본에서 실행해서 부하 분산(load balancing)하는 방법을 설명한다.

Redis Cluster에 접속하는 방법

1) jar 파일 복사: lettuce-core-6.1.7.RELEASE.jar, dependencies/* -> C:\eclipse\workspace2/libs/ 2) Create Project Menu: File -> New -> Java Project Project name: Cluster JRE: javaSE-1.8 Finish 3) 이미 만들어 놓은 lettuce lib 추가 할 때: Project name (우클릭) -> Properties -> Java Build Path -> Tab: Libraris -> Add Library -> User Library -> Next -> lettuce (선택) -> Finish -> Apply and Close 4) Class (Main.java, Util.java, Config.java) 만들기 Menu: File -> New -> Class -> Name: Main, public static void main (선택) -> Finish Menu: File -> New -> Class -> Name: Util -> Finish Menu: File -> New -> Class -> Name: Config -> Finish

Main.java

import java.text.SimpleDateFormat; import java.time.Duration; import java.util.ArrayList; import java.util.Date; import java.util.HashMap; import java.util.List; import java.util.Scanner; import java.util.Set; import io.lettuce.core.KeyValue; import io.lettuce.core.ReadFrom; import io.lettuce.core.RedisFuture; import io.lettuce.core.cluster.ClusterClientOptions; import io.lettuce.core.cluster.ClusterTopologyRefreshOptions; import io.lettuce.core.cluster.ClusterTopologyRefreshOptions.RefreshTrigger; import io.lettuce.core.cluster.RedisClusterClient; import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; public class Main { public static String workDir = ""; public static String clusterConfigFileName = "cluster.cfg"; public static ArrayList<String[]> serverList = new ArrayList<String[]>(); public static RedisClusterClient clusterClient = null; public static StatefulRedisClusterConnection<String, String> connection = null; public static Util util = new Util(); public static SimpleDateFormat format1 = new SimpleDateFormat ( "yyyy-MM-dd HH:mm:ss.SSS"); public static void main(String[] args) { cluster(); } @SuppressWarnings({ "unused", "resource" }) public static void cluster() { System.out.println("Hello RedisCluster-Lettuce World"); System.out.println("Redis-7.0.3, Lettuce-6.1.7"); Main.workDir = System.getProperty("user.dir")+"\\"; System.out.println("workdir: "+Main.workDir); // cluster.cfg 파일에서 server ip,port,password를 읽어온다. Config config = new Config(); config.loadConfig(); String clusterConn = "redis://"; String ip = null; String password = null; for (int i=0; i < Main.serverList.size(); i++) { String conn[] = Main.serverList.get(i); if (i==0) { password = conn[3]; ip = conn[1]; // redis server connection clusterConn += password+"@"+ip+":"+conn[2]; // password, ip:port } else { clusterConn += ","+conn[1]+":"+conn[2]; // ip:port } } clusterConn += "/0?timeout=10s"; System.out.println(clusterConn); // Syntax: redis://[password@]host[:port] // Syntax: redis://[username:password@]host[:port] // Syntax: redis://[password@]host1:port,host2:port,host3:port/0?timeout=60s clusterClient = RedisClusterClient.create(clusterConn); ClusterTopologyRefreshOptions topologyRefreshOptions = null; boolean period = true; if (period) { // ---------------------------------------------------------------------------------------- // Lettuce source code // cluster/ClusterTopologyRefreshOptions.java // public Builder enablePeriodicRefresh(Duration refreshPeriod) { // return refreshPeriod(refreshPeriod).enablePeriodicRefresh(); // } // ---------------------------------------------------------------------------------------- // https://lettuce.io/core/release/reference/#clientoptions.cluster-specific-options // 7.2.1. Cluster-specific options // enablePeriodicRefresh: default 60sec // enablePeriodicRefresh()를 설정하지 않으면 false로 failover 시 재 접속하지 못한다. // connection timeout후에 작동한다. topologyRefreshOptions = ClusterTopologyRefreshOptions.builder() .enablePeriodicRefresh(Duration.ofSeconds(20)) .build(); } else { // RefreshTrigger.ASK_REDIRECT -> Redis responds with a {@code ASK} redirection to a command. // 레디스 명령에 ASK로 응답: slot이 이동 중인 경우 // RefreshTrigger.MOVED_REDIRECT -> Redis responds with a {@code MOVED} redirection to a command. // 레디스 명령에 MOVED로 응답: slot이 이동된 경우(아직 cluster nodes로 새 정보를 가져오지 못한 경우) // RefreshTrigger.PERSISTENT_RECONNECTS -> Connections to a particular host run into persistent reconnects (more than one attempt). // RefreshTrigger.UNCOVERED_SLOT -> Attempts to use a slot that is not covered by a known node. // RefreshTrigger.UNKNOWN_NODE -> Connection attempts to unknown nodes. // adaptiveRefreshTriggersTimeout: default 30sec topologyRefreshOptions = ClusterTopologyRefreshOptions.builder() // .enableAdaptiveRefreshTrigger(RefreshTrigger.PERSISTENT_RECONNECTS) .enableAllAdaptiveRefreshTriggers() // all trigger // .adaptiveRefreshTriggersTimeout(Duration.ofSeconds(30)) // default: 30sec .build(); } clusterClient.setOptions(ClusterClientOptions.builder() .topologyRefreshOptions(topologyRefreshOptions) .build()); connection = clusterClient.connect(); connection.setReadFrom(ReadFrom.ANY); System.out.println("Connected to Redis"); String timeFormat = "["+format1.format(new Date())+"] "; System.out.println("Start: "+timeFormat); // StatefulRedisConnection<String, String> conn7000 = connection.getConnection(ip, 7000); int loop = 1000; Scanner scanner = new Scanner(System.in); String ret = null; while(true) { // 서버별로 명령을 실행할 때는 getConnection(ip, port)를 사용한다. try { ret = connection.getConnection(ip, 7000).sync().configResetstat(); System.out.println("7000 config resetstat -> "+ret); } catch (Exception e) { System.out.println(e.getMessage()); } try { ret = connection.getConnection(ip, 7001).sync().configResetstat(); System.out.println("7001 config resetstat -> "+ret); } catch (Exception e) { System.out.println(e.getMessage()); } try { ret = connection.getConnection(ip, 7002).sync().configResetstat(); System.out.println("7002 config resetstat -> "+ret); } catch (Exception e) { System.out.println(e.getMessage()); } try { ret = connection.getConnection(ip, 8000).sync().configResetstat(); System.out.println("8000 config resetstat -> "+ret); } catch (Exception e) { System.out.println(e.getMessage()); } try { ret = connection.getConnection(ip, 8001).sync().configResetstat(); System.out.println("8001 config resetstat -> "+ret); } catch (Exception e) { System.out.println(e.getMessage()); } try { ret = connection.getConnection(ip, 8002).sync().configResetstat(); System.out.println("8002 config resetstat -> "+ret); } catch (Exception e) { System.out.println(e.getMessage()); } System.out.println("1:SET Sync 2:SET Async\n"+ "3:MSET 4:MGET\n"+ "5:SUNION\n"+ "6:ANY 7:ANY_REPLICA\n"+ "8:MASTER=UPSTREAM 9:MASTER_PREFERRED=UPSTREAM_PREFERRED\n"+ "10:REPLICA 11:REPLICA_PREFERRED\n"+ "12:LOWEST_LATENCY"); System.out.println("선택(exit:99)> "); String input = scanner.nextLine(); int i = util.checkInt(input); switch (i) { // flushdb()는 마스터 노드 전체에 적용된다. // https://lettuce.io/core/release/reference/#redis-cluster.cross-slot-command-routing // 5.3.2. Cross-slot command execution and cluster-wide execution for selected commands // 마스터 노드 전체에 적용되는 명령: dbsize, keys, shutdown // 멀티키 명령 실행 가능: mset, mset -> key 개수만큼 set, get 명령이 실행된다. case 1: /* connection.sync().flushdb(); */ syncSet(loop); break; case 2: /* connection.sync().flushdb(); */ asyncSet(loop); break; case 3: mset(loop); break; case 4: mget(loop); break; case 5: sunion(); break; case 6: connection.setReadFrom(ReadFrom.ANY); break; case 7: connection.setReadFrom(ReadFrom.ANY_REPLICA); break; case 8: connection.setReadFrom(ReadFrom.MASTER); break; case 9: connection.setReadFrom(ReadFrom.MASTER_PREFERRED); break; case 10: connection.setReadFrom(ReadFrom.REPLICA); break; case 11: connection.setReadFrom(ReadFrom.REPLICA_PREFERRED); break; case 12: connection.setReadFrom(ReadFrom.LOWEST_LATENCY); break; case 99: break; default: System.out.println("잘못 선택했습니다. 다시 선택하세요. "); } if (i >= 6 && i <= 12) { if (true) syncGet(loop); if (false) { asyncGet(loop); // 3초 대기 후 info commandstats 명령을 실행한다. util.sleep(3000); } } try { ret = connection.getConnection(ip, 7000).sync().info("commandstats"); System.out.println("7000 info commandstats -> "+ret); } catch (Exception e) { System.out.println(e.getMessage()); } try { ret = connection.getConnection(ip, 7001).sync().info("commandstats"); System.out.println("7001 info commandstats -> "+ret); } catch (Exception e) { System.out.println(e.getMessage()); } try { ret = connection.getConnection(ip, 7002).sync().info("commandstats"); System.out.println("7002 info commandstats -> "+ret); } catch (Exception e) { System.out.println(e.getMessage()); } try { ret = connection.getConnection(ip, 8000).sync().info("commandstats"); System.out.println("8000 info commandstats -> "+ret); } catch (Exception e) { System.out.println(e.getMessage()); } try { ret = connection.getConnection(ip, 8001).sync().info("commandstats"); System.out.println("8001 info commandstats -> "+ret); } catch (Exception e) { System.out.println(e.getMessage()); } try { ret = connection.getConnection(ip, 8002).sync().info("commandstats"); System.out.println("8002 info commandstats -> "+ret); } catch (Exception e) { System.out.println(e.getMessage()); } if (i==99) break; } Long dbsize = connection.sync().dbsize(); System.out.println("dbszie "+dbsize); timeFormat = "["+format1.format(new Date())+"] "; System.out.println("End: "+timeFormat); connection.close(); clusterClient.shutdown(); System.out.println("Close"); } // Sync Set public static void syncSet(int loop) { long time1 = System.currentTimeMillis(); for (int i=1; i<=loop; i++) { String key = "key-"+String.format("%05d", i); String value = "value-"+String.format("%05d", i); try { util.sleep(100); // failover test 시: 100ms 대기 String ret = connection.sync().set(key, value); String timeFormat = "["+format1.format(new Date())+"] "; System.out.println(timeFormat+key+" --> "+ret); } catch (Exception e) { String timeFormat = "["+format1.format(new Date())+"] "; System.out.println(timeFormat+key+" --> Fail"); i--; // 실패했을 경우 1을 빼서 데이터가 유실되지 않고 입력되게 한다. // redis cluster 정보를 새로고침한다. 하지만 master down 1분 후 실행된다. // enablePeriodicRefresh(true) 설정했으면 필요없다. // clusterClient.refreshPartitions(); // System.out.println("refreshPartitions"); continue; } } long time2 = System.currentTimeMillis(); String msg = util.elapse(time1, time2); System.out.println("Sync Set: "+msg); } // Async Set public static void asyncSet(int loop) { long time1 = System.currentTimeMillis(); for (int i=1; i<=loop; i++) { String key = "key-"+String.format("%05d", i); String value = "value-"+String.format("%05d", i); try { connection.async().set(key, value); } catch (Exception e) { String timeFormat = "["+format1.format(new Date())+"] "; System.out.println(timeFormat+key+" --> Fail"); continue; } } long time2 = System.currentTimeMillis(); String msg = util.elapse(time1, time2); System.out.println("Async Set: "+msg); } // MSET public static void mset(int loop) { long time1 = System.currentTimeMillis(); HashMap<String, String> map = new HashMap<>(); for (int i=1; i<=loop; i++) { String key = "keyMSET-"+String.format("%05d", i); String value = "valueMSET-"+String.format("%05d", i); System.out.println(key+" "+value); map.put(key,value); if (i%10 == 0) { try { util.sleep(100); // failover test 시: 100ms 대기 String ret = connection.sync().mset(map); String timeFormat = "["+format1.format(new Date())+"] "; System.out.println(timeFormat+" --> "+ret); } catch (Exception e) { String timeFormat = "["+format1.format(new Date())+"] "; System.out.println(timeFormat+" --> Fail"); i -= 10; } map.clear(); } } long time2 = System.currentTimeMillis(); String msg = util.elapse(time1, time2); System.out.println("Sync MSET: "+msg); } // MGET public static void mget(int loop) { long time1 = System.currentTimeMillis(); String[] keys = new String[10]; int j=0; for (int i=1; i<=loop; i++) { String key = "keyMSET-"+String.format("%05d", i); System.out.println(key); keys[j++] = key; if (i%10 == 0) { try { util.sleep(100); // failover test 시: 100ms 대기 List<KeyValue<String, String>> ret = connection.sync().mget(keys); String timeFormat = "["+format1.format(new Date())+"] "; System.out.println(timeFormat+" --> "+ret.toString()); } catch (Exception e) { String timeFormat = "["+format1.format(new Date())+"] "; System.out.println(timeFormat+" --> Fail"); i -= 10; } j=0; } } long time2 = System.currentTimeMillis(); String msg = util.elapse(time1, time2); System.out.println("Sync MGET: "+msg); } // SUNION public static void sunion() { long time1 = System.currentTimeMillis(); try { Set<String> ret = connection.sync().sunion("myset01","myset02"); String timeFormat = "["+format1.format(new Date())+"] "; System.out.println(timeFormat+" --> "+ret.toString()); } catch (Exception e) { // ERROR: CROSSSLOT Keys in request don't hash to the same slot // info commandstats: rejected_calls=1 System.out.println(e.getMessage()); String timeFormat = "["+format1.format(new Date())+"] "; System.out.println(timeFormat+" --> Fail"); } long time2 = System.currentTimeMillis(); String msg = util.elapse(time1, time2); System.out.println("Sync SUNION: "+msg); } // Sync Get public static void syncGet(int loop) { long time1 = System.currentTimeMillis(); for (int i=1; i<=loop; i++) { String key = "key-"+String.format("%05d", i); try { util.sleep(100); // failover test 시: 100ms 대기 String ret = connection.sync().get(key); String timeFormat = "["+format1.format(new Date())+"] "; System.out.println(timeFormat+key+" --> "+ret); } catch (Exception e) { String timeFormat = "["+format1.format(new Date())+"] "; System.out.println(timeFormat+key+" --> Fail"); i--; // 실패했을 경우 1을 빼서 데이터가 유실되지 않고 입력되게 한다. continue; } } long time2 = System.currentTimeMillis(); String msg = util.elapse(time1, time2); System.out.println("Sync Get: "+msg); } // Async Get public static void asyncGet(int loop) { long time1 = System.currentTimeMillis(); for (int i=1; i<=loop; i++) { try { RedisFuture<String> ret = connection.async().get("key-"+String.format("%05d", i)); ret.thenAccept(r -> { System.out.println("GET -> "+r.toString()); }); } catch (Exception e) { System.out.println(e.getMessage()); return; } } long time2 = System.currentTimeMillis(); String msg = util.elapse(time1, time2); System.out.println("Async Get: "+msg); } }

Util.java

import java.text.DecimalFormat; import java.util.ArrayList; public class Util { public DecimalFormat decFmt = new DecimalFormat("###,###"); public Util () { } public int checkInt(String val) { try { int i = Integer.parseInt(val); return i; } catch (NumberFormatException e) { return Integer.MIN_VALUE; } } public long checkLong(String val) { try { long l = Long.parseLong(val); return l; } catch (NumberFormatException e) { return Long.MIN_VALUE; } } public double checkDouble(String val) { try { double d = Double.parseDouble(val); return d; } catch (NumberFormatException e) { return Double.MIN_VALUE; } } // space(공란)이 여러개 있어도 처리하고 "str1 str2"도 한 단어로 처리한다. // https://code-examples.net/ko/q/335d89 public String[] split(String str) { str += " "; // To detect last token when not quoted... ArrayList<String> strings = new ArrayList<String>(); boolean inQuote = false; StringBuilder sb = new StringBuilder(); for (int i = 0; i < str.length(); i++) { char c = str.charAt(i); if (c == '"' || c == ' ' && !inQuote) { if (c == '"') inQuote = !inQuote; if (!inQuote && sb.length() > 0) { strings.add(sb.toString()); sb.delete(0, sb.length()); } } else sb.append(c); } return strings.toArray(new String[strings.size()]); } public String elapse(long start, long end) { String elapse = ""; int elapseTime = (int) (end-start); if (elapseTime >= 1000) { elapse = this.decFmt.format(elapseTime/1000.0)+"sec"; } else { elapse = this.decFmt.format(elapseTime)+"ms"; } return "[소요시간:"+elapse+"]"; } public void sleep(int ms) { try { Thread.sleep(ms); } catch (InterruptedException e) { System.out.println(e.getMessage()); } } }

Config.java

import java.io.BufferedReader; import java.io.File; import java.io.FileNotFoundException; import java.io.FileReader; import java.io.IOException; public class Config { Util util = new Util(); public Config() { } public void loadConfig() { String fileName = Main.workDir+Main.clusterConfigFileName; File file = new File(fileName); int idx = 1; try (BufferedReader reader = new BufferedReader(new FileReader(file))) { String line; while ((line = reader.readLine()) != null) { // Tab(9), Carriage return(13), Space(32), , #(35), if (line.charAt(0)==9 || line.charAt(0)==13 || line.charAt(0)==32 || line.charAt(0)==35 ) continue; // space, tab이 여러 개 있으면 name.length=1, 바로 new-line이면 length=0 String name[] = util.split(line); if (name.length <= 1) continue; // 유효한 문자가 없으면 바로 다음 라인을 읽는다. System.out.println(idx+":"+line); idx++; Main.serverList.add(name); } } catch (FileNotFoundException e) { System.out.println(e.getMessage()); } catch (IOException e) { System.out.println(e.getMessage()); } } }

Redis Cluster 구성

Redis Cluster 생성

• 서버 준비: 7000, 7001, 7002, 8000, 8001, 8002 6대를 준비한다.
  redis.conf 변경
  • bind 192.168.56.102 127.0.0.1 -::1
  • port 7000
  • daemonize yes
  • pidfile "/var/run/redis_7000.pid"
  • logfile "redis.log"
  • save ""
  • dir "/app/redis/redis-7.0.3/7000"
  • masterauth "password"
  • requirepass "password"
  • appendonly yes
  • cluster-enabled yes
  • cluster-config-file nodes.conf
  • cluster-node-timeout 5000
  • cluster-require-full-coverage no
• 7000~7002는 마스터로, 8000~8002는 복제본으로 사용할 것이다.
• 클러스터 생성: $ src/redis-cli --cluster create 192.168.56.102:7000 192.168.56.102:7001 192.168.56.102:7002 192.168.56.102:8000 192.168.56.102:8001 192.168.56.102:8002 --cluster-replicas 1 -a password

Lettuce Cluster 설정

레디스 서버 리스트 파일

• 파일 명: cluster.cfg
  
• 위치: 이클립스(eclipse) project 폴더 아래에 작성
• 필드 구성:
  • server-name: 구분자, 프로그램 내에서 사용되지는 않는다.
  • ip: 서버 ip
  • port: 서버 port
  • password: password, 첫 번째 줄에만 입력한다.
    redis.conf requirepass, masterauth에 설정했을 경우 사용한다. # server-name ip port password server-1 192.168.56.102 7000 password server-2 192.168.56.102 7001 server-3 192.168.56.102 7002
• Config.java loadConfig()에서 읽어온다. 빈 라인 또는 #으로 시작하는 라인은 처리하지 않는다.

Lettuce - Redis Cluster 연결

Connection timeout은 default 60sec이다. 여기서는 10sec로 설정했다.
Failover 시 connection timeout이 복구 시간에 영향을 미치기 때문에 주의할 필요가 있다.

몇 가지 연결 방법

• redis://[password@]host[:port]
• redis://[username:password@]host[:port]
• redis://[password@]host1:port,host2:port,host3:port/0?timeout=60s String clusterConn = "redis://"; String ip = null; String password = null; for (int i=0; i < Main.serverList.size(); i++) { String conn[] = Main.serverList.get(i); if (i==0) { password = conn[3]; ip = conn[1]; // redis server connection clusterConn += password+"@"+ip+":"+conn[2]; // password, ip:port } else { clusterConn += ","+conn[1]+":"+conn[2]; // ip:port } } clusterConn += "/0?timeout=10s"; clusterClient = RedisClusterClient.create(clusterConn);

장애 발생 시 대비 Redis Cluster 정보를 새로 가져오는 방법

Redis Cluster 정보를 새로 가져오는(새로고침) 데는 ClusterTopologyRefreshOptions를 사용하고, 2가지 방법으로 설정한다.
기본적으로 새로고침을 활성화하지 않는다. 명시적으로 활성화(enable) 시켜야한다.
Redis Cluster 정보를 새로 가져오는 데는 cluster nodes 명령을 실행한다.

Redis Cluster 정보를 새로 가져오는 2가지 방법

• 주기적으로 클러스터 정보를 가져오는 방법 -> enablePeriodicRefresh()
• 이벤트 발생 시 클러스터 정보를 가져오는 방법 -> enableAdaptiveRefreshTrigger()

2종류의 시간은 기본적으로 아래와 같이 설정된다. 이는 변경할 수 있다.
Source code: ClusterTopologyRefreshOptions.java
    refreshPeriod = 60sec DEFAULT_REFRESH_PERIOD
    adaptiveRefreshTimeout = 30sec DEFAULT_ADAPTIVE_REFRESH_TIMEOUT

Lettuce의 설명을 참조하려면 아래 링크를 클릭하세요.
    7.2.1. Cluster-specific options

EnablePeriodicRefresh()

주기적으로 클러스터 정보를 가져온다.

• enablePeriodicRefresh(Duration) 기간을 설정하고 활성화 시킨다.
  시간 설정 방법: enablePeriodicRefresh(Duration.ofSeconds(30)), enablePeriodicRefresh(Duration.ofMinutes(2))
• enablePeriodicRefresh() -> true로 설정, default 60sec로 설정된다.
• enablePeriodicRefresh(true/false)
  -> refreshPeriod(Duration).enablePeriodicRefresh();
• refreshPeriod(Duration) -> 기간만 설정하는 방법, default 60sec

이 기능을 활성화하면 cluster nodes를 포함해서 다음 4가지 명령이 주기적으로 실행된다.
    cluster nodes, info, client setname, hello
이 명령들은 lettuce에서 redis로 연결(connection)해서 실행하고 바로 연결을 끊는다. 그래서 client list로 볼 수 없다. info stats 명령으로 connection이 증가하는 것을 확인할 수 있다.
# Stats total_connections_received:32
code: cluster/topology/DefaultClusterTopologyRefresh.java
    CompletionStage()
        requestTopology() -> CLUSTER NODES
        requestClients() -> INFO CLIENTS
    openConnections()
        connection.async().clientSetname("lettuce#ClusterTopologyRefresh");
        connection.close();

EnableAdaptiveRefreshTrigger()

이벤트 발생 시 클러스터 정보를 가져온다.

• enableAdaptiveRefreshTrigger(): RefreshTrigger 5가지 중 하나 이상을 설정한다.
  code: enableAdaptiveRefreshTrigger(RefreshTrigger.ASK_REDIRECT, RefreshTrigger.MOVED_REDIRECT);
• enableAllAdaptiveRefreshTriggers(): 5가지 모두 설정한다.

RefreshTrigger 5가지

• RefreshTrigger.ASK_REDIRECT: 레디스 명령에 ASK로 응답할 때,
  cluster setslot importing/migrating 후 cluster setslot node로 slot 이동을 확정하기 전에 원래 노드에 명령이 실행되면 ASK ip:port(대상 노드)가 리턴된다.
  code: onAskRedirection()
• RefreshTrigger.MOVED_REDIRECT: 레디스 명령에 MOVED로 응답할 때,
  cluster setslot importing/migrating/node로 slot를 이동했을 경우 원래 노드에 명령이 실행되면 MOVED ip:port(대상 노드)가 리턴된다.
  code: onAskRedirection()
• RefreshTrigger.UNCOVERED_SLOT: 알려진 노드에 포함되지 않은 슬롯을 사용하려고 할 때
  code: onUncoveredSlot(int slot)
• RefreshTrigger.UNKNOWN_NODE: 알 수 없는 노드에 연결을 시도할 때
  code: onUnknownNode()
• RefreshTrigger.PERSISTENT_RECONNECTS: 특정 노드에 2번 이상 연결을 시도할 때
  code: onReconnectAttempt(int attempt)

테스트 방법

아래 번호를 선택해서 테스트한다.
6번부터 12번 까지는 조회 테스트이다. 조회 테스트 결과는 Sentinel과 같다.

1. SET Sync: SET 명령 실행, Failover 테스트에 사용된다.
2. SET Async: SET 명령을 비동기로 실행한다.
3. MSET: MSET 명령 실행, Redis Cluster에서는 multi-key 명령을 실행할 수 없지만 Lettuce에서는 MSET을 키 개수만큼 SET 명령으로 나누어서 실행한다.
4. MGET: MGET 명령 실행, Lettuce에서는 MGET을 키 개수만큼 GET 명령으로 나누어서 실행한다.
5. SUNION: SUNION 명령 실행 -> SUNION 같은 multi-key 명령은 실행할 수 없다.
6. ANY: 마스터, 복제 노드 고르게 조회 분배
7. ANY_REPLICA: 복제 노드에서만 조회. 복제 노드가 여러개 있으면 고르게 분배, 복제 노드가 다운되면 마스터가 살아있어도 실행 불가
8. MASTER=UPSTREAM: 마스터에서만 조회
9. MASTER_PREFERRED=UPSTREAM_PREFERRED: 마스터 위주로 조회
10. REPLICA: 복제 노드에서만 조회, 복제 노드 중 한 곳에 집중되는 경향이 있다.
11. REPLICA_PREFERRED: 기본적으로 복제 노드에서 조회하고, 복제 노드가 다운되었을 때 마스터에서 조회. 복제 노드 중 한 곳에 집중되는 경향이 있다.
12. LOWEST_LATENCY: 대기 시간이 가장 짧은 노드에서 읽도록 설정한다. 복제 노드 중 한 곳에 집중되는 경향이 있다.

입력 중 서버가 다운되었을 경우 장애 조치(failover) 테스트

enablePeriodicRefresh()와 enableAllAdaptiveRefreshTriggers() 두 가지 방법으로 테스트 한다.

enablePeriodicRefresh() 테스트

boolean period = true;
enablePeriodicRefresh(Duration.ofSeconds(20))

• 1번을 선택해서 SET 명령을 실행한다.
• kill -9 pid로 7000번 마스터 서버를 다운시킨다.
• 36분 26초: server down
• 36분 27초: Lettuce에서 server down 감지
• 36분 33초: Lettuce에서 DefaultClusterTopologyRefresh 실행
• 36분 34초: Redis Cluster Failover 완료: Failover election won: I'm the new master.
• 36분 37초: connection timeout
• 36분 47초: command fail
• 36분 53초: Lettuce에서 DefaultClusterTopologyRefresh 실행
• 36분 57초: command ok

서버 다운 후 30초 만에 복구되었다.
cluster-require-full-coverage no로 설정해도 테스트 결과는 같았다.
Lettuce에서는 이 파라미터를 이용하지 않는 것으로 보인다.

다음 테스트를 위해서 복구 진행

• 다운시킨 7000번 서버 시작
  $ src/redis-server 7000/redis.conf
• slave를 master로 변경
  $ src/redis-cli -p 7000 -a password
  > role -> slave
  > cluster failover
  > role -> master

enableAllAdaptiveRefreshTriggers() 테스트

boolean period = false;
enableAllAdaptiveRefreshTriggers()
adaptiveRefreshTriggersTimeout() default 30sec 적용

• 1번을 선택해서 SET 명령을 실행한다.
• kill -9 pid로 7000번 마스터 서버를 다운시킨다.
• 31분 16초: server down
• 31분 16초: Lettuce에서 server down 감지
• 31분 23초: Redis Cluster Failover 완료: Failover election won: I'm the new master.
• 31분 27초: connection timeout
• 31분 27초: Lettuce에서 DefaultClusterTopologyRefresh 실행
• 31분 37초: command ok

서버 다운 후 20초 만에 복구되었다.

Multi-key 명령, Cluster-wide 명령

Multi-key 명령

Redis Cluster에서는 multi-key 명령을 실행할 수 없으나, Lettuce에서는 일부 multi-key 명령을 single-key(단일키) 명령으로 변경해서 실행할 수 있다.

• DEL, UNLINK: 키 개수만큼 DEL, UNLINK 명령 실행, 예) del key1 key2 key3 -> 삭제한 키 개수 리턴
• EXISTS: 키 개수만큼 EXISTS 명령 실행, 예) exists key1 key2 key3 -> 존재하는 키 개수 리턴
• MSET, MGET: 키 개수만큼 SET, GET 명령으로 변경해서 실행

MSET 명령 테스트

MSET 키 10개씩 100번 실행
  Sync Set: [소요시간:12sec] -> 합계 calls 1000
  7000 info commandstats -> cmdstat_mset:calls=328,usec=703,usec_per_call=2.14
  7001 info commandstats -> cmdstat_mset:calls=340,usec=773,usec_per_call=2.27
  7002 info commandstats -> cmdstat_mset:calls=332,usec=749,usec_per_call=2.26

MGET 명령 테스트

MGET 키 10개씩 100번 실행, ReadFrom.ANY
  Sync MGET: [소요시간:12sec] -> 합계 calls 1000, master, replica 고르게 실행
  7000 info commandstats -> cmdstat_mget:calls=157,usec=319,usec_per_call=2.03
  7001 info commandstats -> cmdstat_mget:calls=179,usec=405,usec_per_call=2.26
  7002 info commandstats -> cmdstat_mget:calls=155,usec=315,usec_per_call=2.03
  8000 info commandstats -> cmdstat_mget:calls=161,usec=332,usec_per_call=2.06
  8001 info commandstats -> cmdstat_mget:calls=177,usec=334,usec_per_call=1.89
  8002 info commandstats -> cmdstat_mget:calls=171,usec=358,usec_per_call=2.09

SUNION 명령 테스트

테스트 데이터 입력
7000> sadd myset03 DDD EEE FFF
7001> sadd myset01 AAA BBB CCC
7001> sadd myset02 CCC DDD EEE
7002> sadd myset04 BBB CCC DDD
code:
connection.sync().sunion("myset01","myset02");
-> ERROR: CROSSSLOT Keys in request don't hash to the same slot
-> 키가 저장된 슬롯이 같아야 실행됨.
7001 info commandstats -> cmdstat_sunion:calls=0,usec=0,usec_per_call=0.00,rejected_calls=1

Cluster-wide 명령

일부 명령은 여러 노드에서 실행해서 결과를 제공한다.

• KEYS: 모든 마스터에서 실행
• SCAN: ReadFrom 설정에 따라 마스터 또는 복제 노드에서 실행
• DBSIZE: 모든 마스터에서 실행해서 합계 키 개수를 리턴
• FLUSHDB, FLUSHALL: 모든 마스터에서 실행, 클러스터 내 모든 데이터가 지워짐.
• SHUTDOWN: 클러스터 모든 노드에서 실행
• CLIENT SETNAME: 클러스터 모든 노드에서 실행
• SCRIPT FLUSH/LOAD/KILL: 클러스터 모든 노드에서 실행

DBSIZE 명령 테스트

Long dbsize = connection.sync().dbsize();
System.out.println("dbszie "+dbsize); -> 1000

Lettuce의 multi-key, cluster-wide 설명을 참조

Lettuce의 multi-key, cluster-wide 설명을 참조하려면 아래 링크를 클릭하세요.
5.3.2. Cross-slot command execution and cluster-wide execution for selected commands

부하 분산(load balancing)하는 방법

조회 명령을 마스터/복제 노드에서 실행해서 부하 분산(load balancing)하는 방법

조회 명령을 마스터/복제 노드에서 실행해서 부하 분산(load balancing)하는 방법을 설명한다.
다음은 Lettuce에서 제공하는 조회 명령을 분산 실행하는 방법이다.
ReadFrom에서 선택할 수 있다.

• ANY: 마스터, 복제 노드에서 고르게 조회 명령을 실행한다.
  code: new ReadFromImpl.ReadFromAnyNode(); // @since 5.2
• ANY_REPLICA: 복제 노드에서 조회 명령을 실행한다. 복제 노드가 여러개 있을 경우 고르게 실행된다.
  복제 노드이 없을 경우 명령을 실행할 수 없다.
  code: new ReadFromImpl.ReadFromAnyReplica(); // @since 6.0.1
• MASTER = UPSTREAM: 마스터에서만 조회 명령을 실행한다.
  code: new ReadFromImpl.ReadFromUpstream(); // @since 6.0
• MASTER_PREFERRED = UPSTREAM_PREFERRED: 기본적으로 마스터에서 실행하고 마스터가 사용할 수 없을 경우 복제 노드에서 실행한다.
  code: new ReadFromImpl.ReadFromUpstreamPreferred(); // @since 6.0
• REPLICA = SLAVE: 복제 노드에서만 조회 명령을 실행한다. 복제 노드가 여러개 있을 경우 특정 노드에 집중되는 경향이 있다. 복제 노드이 없을 경우 명령을 실행할 수 없다.
  code: new ReadFromImpl.ReadFromReplica(); // @since 5.2
• REPLICA_PREFERRED = SLAVE_PREFERRED: 기본적으로 복제 노드에서 실행하고 복제 노드을 사용할 수 없을 경우 마스터에서 실행한다. 복제 노드가 여러개 있을 경우 특정 노드에 집중되는 경향이 있다.
  code: new ReadFromImpl.ReadFromReplicaPreferred(); // @since 5.2
• LOWEST_LATENCY = NEAREST: 대기 시간이 가장 짧은 노드에서 읽도록 설정한다. 복제 노드가 여러개 있을 경우 특정 노드에 집중되는 경향이 있다.
  new ReadFromImpl.ReadFromLowestCommandLatency(); // @since 6.1.7

테스트 결과는 Sentinel에서와 같다.

Source code

DefaultClusterTopologyRefresh.java

DefaultClusterTopologyRefresh.java -> implements ClusterTopologyRefresh CompletionStage() requestTopology() -> CLUSTER NODES requestClients() -> INFO CLIENTS openConnections() String message = String.format("Unable to connect to [%s]: %s", socketAddress, throwableToUse.getMessage() != null ? throwableToUse.getMessage() : throwableToUse.toString()); connection.async().clientSetname("lettuce#ClusterTopologyRefresh"); connection.close(); Connections.java requestTopology() -> CLUSTER NODES requestClients() -> INFO CLIENTS TopologyComparators.java sortByUri() sortByClientCount() sortByLatency()