redis clients lettuce2 lettuce2_cluster

Lettuce Redis Cluster

Redis Developer Course	Redis Technical Support	Redis Enterprise Server

Lettuce Redis Cluster

이 문서는 레터스(Lettuce)를 이용해서 레디스 클러스터(Redis Cluster) 용 Java Spring 애플리케이션을 개발하는 방법을 설명합니다.
레디스 클러스터에 연결(접속)하는 방법, 마스터가 다운되는 장애 발생 시 레터스 작동 방식, Pub/Sub과의 관계, Pipeline 등을 설명하고 관련 자바 소스 코드(source code)를 제공합니다.

기본 사용법

① RedisURI redisUri = RedisURI.Builder.redis(host[0],Integer.parseInt(host[1]))

레디스 클러스터 서버의 주소(ip와 port)와 추가 정보로 RedisURI를 생성합니다.

② redisClusterClient = RedisClusterClient.create(nodes);

redisClusterClient 인스턴스를 생성합니다.

③ ClusterTopologyRefreshOptions.builder()

ClusterTopologyRefreshOptions 인스턴스를 생성해서 PeriodicRefresh()를 설정합니다.

④ clusterConnection = redisClusterClient.connect();

레디스 클러스터에 연결합니다.

⑤ clusterConnection.setReadFrom(redisProps.getReadFrom());

조회 명령은 복제서버에서 실행하기 위해 ReadFrom을 설정합니다.
ReadFrom 각 값에 대한 자세한 설명은 여기를 보세요.

⑥ clusterCommands = clusterConnection.sync();

Sync 명령 실행용 API를 얻습니다.

Source code
① List<RedisURI> nodes = new ArrayList<>(); for (String node: redisProps.getClusterNodes()) { String[] host = node.split(":"); RedisURI redisUri = RedisURI.Builder .redis(host[0],Integer.parseInt(host[1])) // IP, Port .withPassword(redisProps.getPassword()) // Redis password 설정 .withTimeout(redisProps.getTimeout()) // Command timeout 설정 .withClientName(redisProps.getClientName()) // Client name 설정 .build(); nodes.add(redisUri); } ② redisClusterClient = RedisClusterClient.create(nodes); ③ ClusterTopologyRefreshOptions topologyRefreshOptions = ClusterTopologyRefreshOptions.builder() .enablePeriodicRefresh(redisProps.getPeriodicRefresh()) // default 60 sec .build(); ④ clusterConnection = redisClusterClient.connect(); ⑤ clusterConnection.setReadFrom(redisProps.getReadFrom()); ⑥ clusterCommands = clusterConnection.sync();

장애복구(failover)와 Application 연결

애플리케이션(application)은 PeriodicRefresh 설정에 따라 주기적(기본값 60초)으로 CLUSTER NODES와 INFO 명령을 실행해서 레디스 클러스터 상태 변화를 감지합니다.
레디스 클러스터는 센티널처럼 publish 메시지를 보내지 않습니다.

다음은 애플리케이션(Application) 시작, 장애 발생(마스터 다운), 장애 복구, 클러스터 정보 조회(ClusterTopologyRefresh), SET 명령 실행을 시간 순으로 정리했습니다.
이것은 Application을 debug mode로 설정해서 얻은 자료입니다.

• 애플리케이션(Application) 시작
  09:34:49: Executing ':MainApplication.main()'...
• 1분에 한번씩 레디스 클러스터 구정 정보를 조회: 명령 CLUSTER NODES, INFO
  09:36:14.526 : ClusterTopologyRefreshScheduler.run()
• http://localhost:8080/set/key01 실행
  redisClusterClient.connect()할 때 레디스 클러스터내 모든 서버에 연결하고 정보를 얻어오고 연결을 끊습니다. 이후 최초 실제 명령 실행 시 연결합니다. 이 연결은 유지됩니다.
  09:36:18.310 : Received [GET /set/key01 HTTP/1.1
  09:36:18.457 : getConnection(WRITE, 13770) -> key01 slot no
  09:36:18.459 : Connecting to Redis at 192.168.56.102:8002
  // 처음 명령을 실행하므로 HELLO로 연결하고 SET 명령을 실행한다.
  09:36:18.466 : [192.168.56.102:8002] write(ctx, AsyncCommand [type=HELLO])
  09:36:18.471 : [192.168.56.102:8002] write(ctx, [command=AsyncCommand [type=SET]])
  09:36:18.514 : Completed 200 OK
• 1분에 한번씩 레디스 클러스터 구정 정보를 조회: 명령 CLUSTER NODES, INFO
  09:37:14.525 : ClusterTopologyRefreshScheduler.run()
• 마스터-1 다운
  09:37:22.414 - shutdown
• http://localhost:8080/set/key01 실행 -> 다운되지 않은 노드는 정상적으로 실행된다.
  09:37:51.412 : Received [GET /set/key01 HTTP/1.1
  09:37:51.418 : getConnection(WRITE, 13770)
  // 위에서 연결했으므로 SET 명령만 실행하면 된다.
  09:37:51.419 : [192.168.56.102:8002] write() [command=AsyncCommand [type=SET]]
  09:37:51.426 Completed 200 OK
• http://localhost:8080/set/key02 실행 -> 다운된 노드는 에러 발생
  09:37:54.696 : Received [GET /set/key02 HTTP/1.1
  09:37:54.701 : getConnection(WRITE, 1449)
  // Master-1(8000)은 다운되었으므로 에러 발생
  09:37:54.702 : Connecting to Redis at 192.168.56.102:8000
  Connection refused: no further information: /192.168.56.102:8000
• 장애복구 완료: cluster-node-timeout 30000 (30초)
  09:37:56.893 * Replica -> Master
• http://localhost:8080/set/key03 실행
  09:38:08.806 : Received [GET /set/key03 HTTP/1.1
  09:38:08.809 : getConnection(WRITE, 5512)
  // 처음 명령을 실행하므로 HELLO로 연결하고 SET 명령을 실행한다.
  09:38:08.810 : Connecting to Redis at 192.168.56.102:8001
  09:38:08.813 : [192.168.56.102:8001] write(ctx, AsyncCommand [type=HELLO])
  09:38:08.816 : [192.168.56.102:8001] write(ctx, [command=AsyncCommand [type=SET]])
  09:38:08.821 Completed 200 OK
• http://localhost:8080/set/key04 실행
  09:38:12.418 : Received [GET /set/key04 HTTP/1.1
  // 위에서 연결했으므로 SET 명령만 실행하면 된다.
  09:38:12.421 : getConnection(WRITE, 9583)
  09:38:12.421 : [192.168.56.102:8001] write(ctx, [command=AsyncCommand [type=SET]])
  09:38:12.425 : Completed 200 OK
• 1분에 한번씩 레디스 클러스터 구정 정보를 조회: 명령 CLUSTER NODES, INFO
  -> 클러스터 정보 새로 고침, 새 마스터 인지
  09:38:14.526 : ClusterTopologyRefreshScheduler.run()
• http://localhost:8080/set/key05 실행
  09:38:16.341 : Received [GET /set/key05 HTTP/1.1
  09:38:16.344 : getConnection(WRITE, 13646)
  09:38:16.344 : [192.168.56.102:8002] write(ctx, [command=AsyncCommand [type=SET]])
  09:38:16.352 : Completed 200 OK
• http://localhost:8080/set/key06 실행
  09:38:21.325 : Received [GET /set/key06 HTTP/1.1
  09:38:21.328 : getConnection(WRITE, 1325)
  09:38:21.329 : Connecting to Redis at 192.168.56.102:8003
  09:38:21.331 : [192.168.56.102:8003] write(ctx, AsyncCommand [type=HELLO])
  09:38:21.335 : [192.168.56.102:8003] write(ctx, [command=AsyncCommand [type=SET]])
  09:38:21.340 Completed 200 OK

레디스 클러스터와 파이프라인(pipeline)

Pipeline은 일련의 명령을 모아서 한번에 레디스 서버로 보내서 실행하는 것이다. 그러므로 레디스 클러스터에서는 되지 않을거 같지만 된다.
Lettuce에서 키들을 서버별로 분류해서 각 서버에 한번에 보내서 실행한다.
• Pipeline에 대한 설명은 여기를 보세요.

• 애플리케이션(Application) 시작
  14:06:30: Executing ':MainApplication.main()'...
• SET 명령 10개를 pipeline으로 보내는 메서드 실행
  다음은 debug 모드로 애플리케이션을 실행해서 얻은 데이터이다. 각 명령마다 해시값을 추출해서 각 서버에 할당한다.
  http://localhost:8080/pipeline/key -> 명령 실행
  14:08:44.239 : Received [GET /pipeline/key HTTP/1.1
  14:08:44.352 : dispatching command AsyncCommand [type=SET]
  14:08:44.353 : getConnection(WRITE, 15130)
  14:08:44.360 : dispatching command AsyncCommand [type=SET]
  14:08:44.360 : getConnection(WRITE, 11067)
  14:08:44.361 : dispatching command AsyncCommand [type=SET]
  14:08:44.361 : getConnection(WRITE, 7000)
  14:08:44.364 : dispatching command AsyncCommand [type=SET]
  14:08:44.364 : getConnection(WRITE, 2937)
  14:08:44.366 : dispatching command AsyncCommand [type=SET]
  14:08:44.366 : getConnection(WRITE, 15262)
  14:08:44.366 : dispatching command AsyncCommand [type=SET]
  14:08:44.366 : getConnection(WRITE, 11199)
  14:08:44.367 : dispatching command AsyncCommand [type=SET]
  14:08:44.367 : getConnection(WRITE, 7132)
  14:08:44.367 : dispatching command AsyncCommand [type=SET]
  14:08:44.367 : getConnection(WRITE, 3069)
  14:08:44.367 : dispatching command AsyncCommand [type=SET]
  14:08:44.367 : getConnection(WRITE, 14866)
  14:08:44.367 : dispatching command AsyncCommand [type=SET]
  14:08:44.367 : getConnection(WRITE, 10803)
• 각 서버별로 모아진 명령을 보낸다.
  Master-1(8000) 서버에는 2개, Master-2(8001) 서버에는 3개, Master-3(8002) 서버에는 5개 명령을 보낸다.
  14:08:44.372 : [192.168.56.102:8000] flushCommands() Flushing 2 commands
  14:08:44.368 : [192.168.56.102:8001] flushCommands() Flushing 3 commands
  14:08:44.373 : [192.168.56.102:8002] flushCommands() Flushing 5 commands
  
• Master-1(8000) 서버 로그: 2개 명령이 실행되었다.
  1598:M 14:08:44.451 - Accepted 192.168.56.1:12513
  1598:M 14:08:44.452 - 192.168.56.1:12513: HELLO 3 AUTH default redisgate SETNAME redisgate
  1598:M 14:08:44.459 - 192.168.56.1:12513: SET key-00003 value-00003
  1598:M 14:08:44.460 - 192.168.56.1:12513: SET key-00007 value-00007
• Master-2(8001) 서버 로그: 3개 명령이 실행되었다.
  1605:M 14:08:44.449 - Accepted 192.168.56.1:12512
  1605:M 14:08:44.449 - 192.168.56.1:12512: HELLO 3 AUTH default redisgate SETNAME redisgate
  1605:M 14:08:44.460 - 192.168.56.1:12512: SET key-00002 value-00002
  1605:M 14:08:44.461 - 192.168.56.1:12512: SET key-00006 value-00006
  1605:M 14:08:44.461 - 192.168.56.1:12512: SET key-00009 value-00009
• Master-3(8002) 서버 로그: 5개 명령이 실행되었다.
  Master-3(8002)
  1612:M 14:08:44.443 - Accepted 192.168.56.1:12511
  1612:M 14:08:44.444 - 192.168.56.1:12511: HELLO 3 AUTH default redisgate SETNAME redisgate
  1612:M 14:08:44.462 - 192.168.56.1:12511: SET key-00000 value-00000
  1612:M 14:08:44.462 - 192.168.56.1:12511: SET key-00001 value-00001
  1612:M 14:08:44.462 - 192.168.56.1:12511: SET key-00004 value-00004
  1612:M 14:08:44.462 - 192.168.56.1:12511: SET key-00005 value-00005
  1612:M 14:08:44.462 - 192.168.56.1:12511: SET key-00008 value-00008

레디스 클러스터와 Pub/Sub과의 관계

레디스 클러스터에서 Pub/Sub을 사용하려면 "redisClusterClient.connectPubSub();"를 사용하세요.
Publish와 Subscribe 용 연결(connection)을 구분해서 사용하시기 바랍니다.
Subscribe 명령을 실행하면 레터스(lettuce)는 여러 마스터 중 하나에 명령을 실행합니다. 해당 마스터가 다운되면 레터스(lettuce)는 바로 살아있는 다른 마스터에 Subscribe 명령을 실행합니다.
그래서, Pub/Sub 기능을 중단없이 사용할 수 있습니다.
예) 이전 마스터에 ch01 ~ ch05까지 5번 SUBSCRIBE를 했다면 새 마스터에 이렇게 실행합니다.
    SUBSCRIBE ch05 ch03 ch04 ch01 ch02
• Pub/Sub에 대한 소개와 사용법은 여기를 보세요.

Connect String(URI)을 이용한 연결(접속) 방법

Redis URI syntax
redis :// [[username :] password@] host [:port][/database] [?[timeout=timeout[d|h|m|s|ms|us|ns]] [&clientName=clientName] [&libraryName=libraryName] [&libraryVersion=libraryVersion] ]
아래와 같은 방법으로 Connect String(URI)을 만들어서 레디스 클러스터에 연결한다. List<RedisURI> nodes = new ArrayList<>(); for(String node: redisProps.getClusterNodes()) { String uri = "redis://redisgate@"+node+"?&clientName="+redisProps.getClientName() +"&timeout="+redisProps.getTimeout(); System.out.println(uri); nodes.add(RedisURI.create(uri)); } redisClusterClient = RedisClusterClient.create(nodes);

Lettuce Spring Project

Spring Project 생성

이제 Java Spring Project를 생성해서 애플리케이션을 작성/실행해 봅니다.
Java Spring Project 생성 URL: start.spring.io

• Project: Gradle-Groovy
• Language: Java
• Spring Boot: 3.1.8
• Project Metadata
  • Group: com.redisgate
  • Artifact: ClusterLettuce -> Project 명칭, 이 이름으로 압축파일이 생긴다.
  • Name: Main -> Main Class Name
  • Description: Redis Cluster Lettuce Application -> 설명
  • Package name: com.redisgate.redis -> Package Name
  • Packaging: Jar
  • Java: 17 -> Java 버전 선택
• Dependencies에서 [ADD … CTRL + B] 버튼을 클릭해서 아래 3개를 추가합니다.
  • Spring Web
  • Lombok
  • Spring Data Redis(Access+Driver)
• 마지막으로 [GENERATE CTRL + ⏎ ] 버튼 클릭해서 압축 파일(zip)을 다운받는다.
• 스프링부트3.x 는 자바17 이상, 스프링부트 2.x는 자바11을 사용합니다.

이 프로젝트에 사용된 Lettuce, Spring, Redis 버전

• Lettuce-6.2.7
• Spring-Boot: 3.1.8, Spring: 6.0.16
• Redis-7.2.3

Class/File 구성

• Main Class: 메인 클래스
• application.properties: 레디스 클러스터 연결 값들
• RedisProps class: application.properties 파일을 읽어오는 클래스
• RedisConn class: 레디스 클러스터에 연결하는 클래스
• RedisController class: 레디스 명령 테스트 클래스
• RedisPipeline class: Pipeline 테스트 클래스
• RedisPubSub class: Pub/Sub 테스트 클래스

Main Class

package com.redisgate.redis; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class MainApplication { public static void main(String[] args) { SpringApplication.run(MainApplication.class, args); System.out.println("----- Redis Cluster Lettuce Application 시작 -----"); } }

application.properties

# Redis Cluster Lettuce Properties redis.clusterNodes: 192.168.56.102:8000, 192.168.56.102:8001, 192.168.56.102:8002, 192.168.56.102:8003, 192.168.56.102:8004, 192.168.56.102:8005 redis.password: redisgate redis.timeout: 5s redis.connectTimeout: 5s redis.clientName: redisgate redis.readFrom: replicaPreferred redis.periodicRefresh: 60s #server.port: 9090 #logging.level.root: debug

RedisProps class

application.properties 파일을 읽어오는 클래스.

• clusterNodes는 레디스 클러스터 노드 주소(ip:port)를 읽어옵니다. List를 사용했습니다.
• password는 char[]를 사용합니다. String을 사용하면 적용 시 getPassword().toCharArray()를 사용해야 합니다.
• getReadFrom()은 readFrom 문자열을 ReadFrom 값으로 변경합니다.
• periodicRefresh는 레디스 클러스터내 레디스 서버에 주기적으로 접속해서 CLUSTER NODES와 INFO 명령을 실행해서 구성 정보를 조회합니다. 기본값은 60초이고 여기서는 application.properties에서 설정하도록 했습니다.

package com.redisgate.redis; import io.lettuce.core.ReadFrom; import lombok.Getter; import lombok.Setter; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Configuration; import java.time.Duration; import java.util.List; @Configuration @ConfigurationProperties(prefix = "redis") @Getter @Setter public class RedisProps { private List<String> clusterNodes; private char[] password; private Duration timeout; private Duration connectTimeout; private String clientName; private String readFrom; private Duration periodicRefresh; /** * 조회 명령을 복제서버에 실행하도록하는 ReadFrom * application.properties에 입력한 문자열(String)을 ReadFrom 값으로 만든다. * 자세한 설명은 아래 URL을 참고하세요. * http://redisgate.kr/redis/clients/lettuce2_readfrom.php * @return ReadFrom */ public ReadFrom getReadFrom() { return ReadFrom.valueOf(readFrom); } /** * application.properties에 입력한 문자열(String)을 리턴한다. * @return String */ public String getReadFromString() { return readFrom; } }

RedisConn class

레디스 클러스터에 연결합니다.
topologyRefreshOptions과 connectTimeout을 설정을 합니다.

다음은 connect 메서드 실행 시 레디스 서버에 실행되는 명령입니다. 마스터, 복제 서버 모두에 동일한 명령이 실행됩니다.

• redisClusterClient.connect()
  
  Redis Server
      HELLO 3 AUTH default redisgate SETNAME redisgate
      CLIENT SETNAME lettuce#ClusterTopologyRefresh
      CLUSTER NODES
      INFO
      closed connection

package com.redisgate.redis; import io.lettuce.core.RedisURI; import io.lettuce.core.SocketOptions; 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.RedisAdvancedClusterCommands; import io.lettuce.core.cluster.pubsub.StatefulRedisClusterPubSubConnection; import io.lettuce.core.cluster.pubsub.api.sync.RedisClusterPubSubCommands; import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.springframework.context.annotation.Configuration; import java.util.ArrayList; import java.util.List; @Configuration @Getter @Slf4j public class RedisConn { private final RedisClusterClient redisClusterClient; private final StatefulRedisClusterConnection<String, String> clusterConn; private final RedisAdvancedClusterCommands<String, String> clusterCommands; private final StatefulRedisClusterConnection<String, String> clusterPipelineConn; private final RedisAdvancedClusterAsyncCommands<String, String> clusterPipelineCommands; private final StatefulRedisClusterPubSubConnection<String, String> clusterPubConn; private final RedisClusterPubSubCommands<String, String> clusterPubCommands; private final StatefulRedisClusterPubSubConnection<String, String> clusterSubConn; private final RedisClusterPubSubCommands<String, String> clusterSubCommands; public RedisConn(RedisProps redisProps) { List<RedisURI> nodes = new ArrayList<>(); for (String node: redisProps.getClusterNodes()) { String[] host = node.split(":"); RedisURI redisUri = RedisURI.Builder .redis(host[0],Integer.parseInt(host[1])) // IP, Port .withPassword(redisProps.getPassword()) // Redis password 설정 .withTimeout(redisProps.getTimeout()) // Command timeout 설정 .withClientName(redisProps.getClientName()) // Client name 설정 .build(); nodes.add(redisUri); } // URI syntax // redis :// [[username :] password@] host [:port][/database] // [?[timeout=timeout[d|h|m|s|ms|us|ns]] [&clientName=clientName] /* URI syntax를 사용하는 방법 for(String node: redisProps.getClusterNodes()) { String uri = "redis://redisgate@"+node+"?&clientName="+redisProps.getClientName() +"&timeout="+redisProps.getTimeout(); System.out.println(uri); nodes.add(RedisURI.create(uri)); } */ redisClusterClient = RedisClusterClient.create(nodes); // 주기적(default 60s)으로 Redis Cluster에 접속해서 구성정보를 조회한다. ClusterTopologyRefreshOptions topologyRefreshOptions = ClusterTopologyRefreshOptions.builder() .enablePeriodicRefresh(redisProps.getPeriodicRefresh()) // default 60 sec .build(); // topologyRefreshOptions과 connectTimeout을 설정을 한다. redisClusterClient.setOptions(ClusterClientOptions.builder() .topologyRefreshOptions(topologyRefreshOptions) .autoReconnect(true) // default true: 1,2,4,8,16,30초, 이후 30초마다 반복해서 재연결 시도 .socketOptions(SocketOptions.builder().connectTimeout(redisProps.getConnectTimeout()).build()) .build()); clusterConnection = redisClusterClient.connect(); clusterConnection.setReadFrom(redisProps.getReadFrom()); clusterCommands = clusterConnection.sync(); // Pipeline 연결 clusterPipelineConn = redisClusterClient.connect(); clusterPipelineCommands = clusterPipelineConn.async(); // Pub/Sub 연결 clusterPubConn = redisClusterClient.connectPubSub(); clusterPubCommands = clusterPubConn.sync(); clusterSubConn = redisClusterClient.connectPubSub(); clusterSubCommands = clusterSubConn.sync(); } }

RedisController class

레디스 클러스터, 레디스 명령 테스트 클래스

1. CLUSTER NODES 클러스터 구성정보를 조회합니다. 임의의 마스터에서 실행됩니다.
   http://localhost:8080/clusterNodes
2. INFO 서버 정보를 조회합니다. 임의의 마스터에서 실행됩니다.
   http://localhost:8080/info
3. ROLE 서버 역할을 조회합니다. 임의의 마스터에서 실행됩니다.
   http://localhost:8080/role
4. PING 임의의 마스터에서 실행됩니다.
   http://localhost:8080/ping
5. SET 키-값 저장
   http://localhost:8080/set/key01
6. GET 조회, 복제에서 읽는지 readFrom 설정 확인
   http://localhost:8080/get/key01
   GET 명령이 복제서버에 실행되는 것을 확인하기 위해서 아래와 같이 redis-cli를 실행 후 위 get 명령을 실행합니다.
   $ bin/redis-cli -p 8003 -a redisgate --stat

package com.redisgate.redis; import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands; import lombok.extern.slf4j.Slf4j; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RestController; import java.util.List; @RestController @Slf4j public class RedisController { private final RedisAdvancedClusterCommands<String, String> clusterCommands; private final StringRedisTemplate stringRedisTemplate; public RedisController(RedisConn redisConn, StringRedisTemplate stringRedisTemplate) { clusterCommands = redisConn.getClusterCommands(); this.stringRedisTemplate = stringRedisTemplate; } // 예제 1) CLUSTERNODES 클러스터 구성정보를 조회한다. // http://localhost:8080/clusterNodes @GetMapping("/clusterNodes") public String clusterNodes() { String msg = "예제 1) clusterNodes -> "; String result = clusterCommands.clusterNodes(); msg += result; System.out.println(msg); return msg; } // 예제 2) INFO 서버 정보를 조회한다. // http://localhost:8080/info @GetMapping("/info") public String info() { String msg = "예제 2) info -> "; String result = clusterCommands.info(); msg += result; System.out.println(msg); return msg; } // 예제 3) ROLE 서버 정보를 조회한다. // http://localhost:8080/role @GetMapping("/role") public String role() { String msg = "예제 2) role -> "; List<Object> result = clusterCommands.role(); msg += result; System.out.println(msg); return msg; } // 예제 4) PING // http://localhost:8080/ping @GetMapping("/ping") public String ping() { String msg = "예제 4) PING -> "; String result = clusterCommands.ping(); msg += result; System.out.println(msg); return msg; } // 예제 5) SET: 키-값 저장 // SET key value [NX|XX] [EX 초] [PX 밀리초] [EXAT timestamp] [PXAT milliseconds-timestamp] [KEEPTTL] [GET] // http://localhost:8080/set/key01 @GetMapping("/set/{key}") public String set(@PathVariable("key") String key) { String msg = "예제 5) SET -> "; String result = clusterCommands.set(key,"value-"+key); msg += result; System.out.println(msg); return msg; } // 예제 6) GET: 조회, 복제에서 읽는지 readFrom 설정 확인 // $ bin/redis-cli -p 8003 -a redisgate --stat // $ bin/redis-cli -p 8004 -a redisgate --stat // $ bin/redis-cli -p 8005 -a redisgate --stat // GET key // http://localhost:8080/get/key01 @GetMapping("/get/{key}") public String get(@PathVariable("key") String key) { String msg = "예제 6) GET -> "; int count = 1; // count를 변경해서 테스트해 보세요. for (int i=0; i < count; i++ ) { clusterCommands.get(key); } msg += count; System.out.println(msg); return msg; } // stringRedisTemplate Test: Lettuce만 설정해도 stringRedisTemplate를 사용할 수 있는지? // Application debug mode: stringRedisTemplate이 생성되는 것을 확인할 수 있다. // o.s.b.f.s.DefaultListableBeanFactory : Creating shared instance of singleton bean 'stringRedisTemplate' // o.s.b.f.s.DefaultListableBeanFactory : Autowiring by type from bean name 'stringRedisTemplate' via factory method to bean named 'redisConnectionFactory' // 실행해보니 localhost:6379로 접속한다. 기본값으로 접속하는 것이다. // 그러므로 Lettuce에 설정된 값을 사용하는 것은 아니다. // io.lettuce.core.AbstractRedisClient : Connecting to Redis at localhost:6379 // http://localhost:8080/set2/keyXX-01 -> Whitelabel Error Page @GetMapping("/set2/{key}") public String set2(@PathVariable("key") String key) { String msg = "예제 1) set2(stringRedisTemplate) -> "+key; stringRedisTemplate.opsForValue().set(key,"valueXX-"+key); System.out.println(msg); return msg; } }

RedisPipeline class

레디스 클러스터 파이프라인(Pipeline) 테스트 클래스

1. Pipeline 레디스 클러스터 Pipeline 테스트
   http://localhost:8080/pipeline/key

package com.redisgate.redis; import io.lettuce.core.LettuceFutures; import io.lettuce.core.RedisFuture; import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands; import lombok.extern.slf4j.Slf4j; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RestController; import java.sql.Timestamp; import java.util.ArrayList; import java.util.List; import java.util.concurrent.TimeUnit; @RestController @Slf4j public class RedisPipeline { private final StatefulRedisClusterConnection<String, String> clusterPipelineConn; private final RedisAdvancedClusterAsyncCommands<String, String> clusterPipelineCommands; public RedisPipeline(RedisConn redisConn) { // Pipeline 용 Async connection을 별도로 만들어서 사용한다. clusterPipelineConn = redisConn.getClusterPipelineConn(); // disable auto-flushing clusterPipelineConn.setAutoFlushCommands(false); clusterPipelineCommands = redisConn.getClusterPipelineCommands(); } // 예제 1) Pipeline 레디스 클러스터 Pipeline 테스트 // http://localhost:8080/pipeline/key @GetMapping("/pipeline/{key}") public String pipeline1(@PathVariable("key") String key) { String msg = "예제 1) Pipeline -> "; System.out.println(msg); Timestamp startTime = new Timestamp(System.currentTimeMillis()); System.out.println("시작: "+startTime); List<RedisFuture<?>> futures = new ArrayList<>(); int iterations = 10; for (int i = 0; i < iterations; i++) { futures.add(clusterPipelineCommands.set(key+String.format("-%05d",i), "value-"+String.format("%05d",i))); } // write all commands to the transport layer clusterPipelineConn.flushCommands(); // synchronization example: Wait until all futures complete boolean result = LettuceFutures.awaitAll(5, TimeUnit.SECONDS, futures.toArray(new RedisFuture[futures.size()])); String resultMsg = "Pipeline("+iterations+") "+key+" -> "+result; System.out.println(resultMsg); Timestamp endTime = new Timestamp(System.currentTimeMillis()); msg += "종료: "+endTime+", 소요시간: "+(endTime.getTime()-startTime.getTime())+"ms"; System.out.println(msg); return msg; } }

RedisPubSub class

Publish/Subscribe 테스트 클래스

1. PubSub Connection에 Listener를 등록
   http://localhost:8080/pubsubAddListener
2. Subscribe 테스트
   http://localhost:8080/subscribe/ch01
3. Publish 테스트
   http://localhost:8080/publish/ch01:Hello

package com.redisgate.redis; import io.lettuce.core.pubsub.RedisPubSubAdapter; import io.lettuce.core.cluster.pubsub.StatefulRedisClusterPubSubConnection; import io.lettuce.core.cluster.pubsub.api.sync.RedisClusterPubSubCommands; import lombok.extern.slf4j.Slf4j; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.bind.annotation.PathVariable; @RestController @Slf4j public class RedisPubSub { private final RedisClusterPubSubCommands<String, String> clusterPubCommands; private final StatefulRedisClusterPubSubConnection<String, String> clusterSubConn; private final RedisClusterPubSubCommands<String, String> clusterSubCommands; public RedisPubSub(RedisConn redisConn) { // Publish commands clusterPubCommands = redisConn.getClusterPubCommands(); // Subscribe commands clusterSubConn = redisConn.getClusterSubConn(); clusterSubCommands = redisConn.getClusterSubCommands(); } // PubSub Connection에 Listener를 등록한다. // 여기서 subscribe 한다. // http://localhost:8080/pubsubAddListener @GetMapping("/pubsubAddListener") public String pubsubAddListener() { // PubSub 메시지 수신 clusterSubConn.addListener(new RedisPubSubAdapter<String, String>() { // 이 메서드는 Publish 명령 실행 시 실행된다. // Channel으로 수신한다. @Override public void message(String channel, String message) { System.out.println("Subscribe(channel) "+channel+" "+message); } // 이 메서드는 Publish 명령 실행 시 실행된다. // Pattern으로 수신한다. @Override public void message(String pattern, String channel, String message) { System.out.println("Subscribe(pattern) "+pattern+" "+channel+" "+message); } }); String msg = "OK - pubsubAddListener"; System.out.println(msg); return msg; } // Sync Subscribe // SUBSCRIBE channel [channel ...] // http://localhost:8080/subscribe/ch01 @GetMapping("/subscribe/{channel}") public String subscribe(@PathVariable("channel") String channel) { try { clusterSubCommands.subscribe(channel); } catch (Exception e) { System.out.println(e.getMessage()); return "Error - subscribe "+channel; } return "OK - subscribe "+channel; } /** * Publish channel message * @param channelMsg channel:message로 구성됩니다. * @return "Publish channel message" */ // PUBLISH channel message // http://localhost:8080/publish/ch01:Hello @GetMapping("/publish/{channelMsg}") public String publish(@PathVariable("channelMsg") String channelMsg) { String[] msg = channelMsg.split(":"); String result = "Publish "+msg[0]+" "+msg[1]; System.out.println(result); Long num = clusterPubCommands.publish(msg[0], msg[1]); return result+" -> "+num; } }