redis clients lettuce2 lettuce2_streams

Lettuce Streams

Redis Developer Course	Redis Technical Support	Redis Enterprise Server

Lettuce Streams

Java Lettuce를 사용한 레디스 스트림(Streams) 명령 사용법입니다.

Streams 소스

Redis6_Stream.java
package com.redisgate.redis; import io.lettuce.core.*; import io.lettuce.core.api.sync.RedisCommands; 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.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; @RestController @Slf4j public class Redis6_Stream { private final RedisCommands<String, String> redisCommands; public Redis6_Stream(RedisConfig redisConfig) { redisCommands = redisConfig.getRedisSyncCommands(); } // 예제 1) XADD: 스트림에 데이터를 추가 // XADD key ID field value [field2 value2 ...] // XADD key [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT count]] *|ID field value [field value ...] // http://localhost:8080/xadd/sensor-1234 @GetMapping("/xadd/{key}") public String xadd(@PathVariable("key") String key) { String msg = "예제 1) XADD -> "; System.out.println(msg); // String xadd(K key, Object... keysAndValues); // String xadd(K key, Map<K, V> body); // String xadd(K key, XAddArgs args, Object... keysAndValues); // String xadd(K key, XAddArgs args, Map<K, V> body); String id; // 온도 입력 id = redisCommands.xadd(key,"temperature", "98.7"); System.out.println(id); // 온도, 습도 입력 id = redisCommands.xadd(key,"temperature", "97.7", "humidity", "50.5"); System.out.println(id); // Map으로 온도, 습도, 압력 입력 Map<String, String> data = new HashMap<>(); data.put("temperature", "90.5"); data.put("humidity", "50.5"); data.put("pressure", "10.5"); id = redisCommands.xadd(key,data); System.out.println(id); return msg+id; } // 예제 2) XRANGE: 데이터 조회 // XRANGE key start end [COUNT count], XREVRANGE key end start [COUNT count] // http://localhost:8080/xrange/sensor-1234 @GetMapping("/xrange/{key}") public String xrange(@PathVariable("key") String key) { String msg = "예제 2) XRANGE -> 전체 조회"; // List<StreamMessage<K, V>> xrange(K key, Range<String> range); // List<StreamMessage<K, V>> xrange(K key, Range<String> range, Limit limit); // Range<String> range = Range.unbounded(); // Range<Double> range = Range.create(2d,5d); // Range<Double> range2 = Range.from(Range.Boundary.excluding(2d),Range.Boundary.excluding(5d)); System.out.println(msg); List<StreamMessage<String, String>> result; result = redisCommands.xrange(key, Range.unbounded()); for (StreamMessage<String, String> item : result) { System.out.println("ID: "+item.getId()); Set<Map.Entry<String, String>> entries = item.getBody().entrySet(); for (Map.Entry<String, String> entry: entries) { System.out.println(" "+entry.getKey()+" "+entry.getValue()); } } msg = "예제 2) XRANGE -> 범위 지정"; System.out.println(msg); result = redisCommands.xrange(key, Range.create("1704983736200","1704983789747")); for (StreamMessage<String, String> item : result) { System.out.println("ID: "+item.getId()); Set<Map.Entry<String, String>> entries = item.getBody().entrySet(); for (Map.Entry<String, String> entry: entries) { System.out.println(" "+entry.getKey()+" "+entry.getValue()); } } return msg+"OK"; } // 예제 3) XLEN: 스트림의 엔트리(ID) 개수 리턴 // XLEN key // http://localhost:8080/xlen/sensor-1234 @GetMapping("/xlen/{key}") public String xlen(@PathVariable("key") String key) { String msg = "예제 3) XLEN -> "; Long result = redisCommands.xlen(key); msg += result; System.out.println(msg); return msg; } // 예제 4) XREAD: 데이터 읽기 // XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...] // http://localhost:8080/xread/sensor-1234:0 -> 처음부터 읽을 때는 0을 지정 // http://localhost:8080/xread/sensor-1234:1704983789750 -> 지정한 ID 이후 데이터를 읽어온다. // http://localhost:8080/xread/sensor-1234:> -> '>' 는 XREADGROUP 명령에서만 사용할 수 있다. @GetMapping("/xread/{key}") @SuppressWarnings("unchecked") public String xread(@PathVariable("key") String key) { String msg = "예제 4) XREAD -> "; // List<StreamMessage<K, V>> xread(StreamOffset<K>... streams); // List<StreamMessage<K, V>> xread(XReadArgs args, StreamOffset<K>... streams); String[] keyId = key.split(":"); // key와 ID를 지정한다. 지정된 ID 이후 데이터를 읽어온다. XReadArgs.StreamOffset<String> stream = XReadArgs.StreamOffset.from(keyId[0],keyId[1]); // 읽어올 데이터 개수를 지정한다. 지정하지 않으면 모든 데이터를 읽어온다. XReadArgs args = XReadArgs.Builder.count(1); List<StreamMessage<String, String>> result = redisCommands.xread(args, stream); msg += result.size(); System.out.println(msg); for (StreamMessage<String, String> item : result) { System.out.println("ID: "+item.getId()); Set<Map.Entry<String, String>> entries = item.getBody().entrySet(); for (Map.Entry<String, String> entry: entries) { System.out.println(" "+entry.getKey()+" "+entry.getValue()); } } return msg; } // 예제 5) XDEL: 데이터 삭제 // XDEL key ID [ID ...] // http://localhost:8080/xdel/sensor-1234:1704983736197 @GetMapping("/xdel/{key}") public String xdel(@PathVariable("key") String key) { String msg = "예제 5) XDEL -> "; String[] keyId = key.split(":"); // 삭제한 id 개수를 리턴한다. Long result = redisCommands.xdel(keyId[0], keyId[1]); msg += result; System.out.println(msg); return msg; } }

각 명령(메서드) 별 표시

XADD

// 예제 1) XADD: 스트림에 데이터를 추가 // XADD key ID field value [field2 value2 ...] // XADD key [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT count]] *|ID field value [field value ...] // http://localhost:8080/xadd/sensor-1234 @GetMapping("/xadd/{key}") public String xadd(@PathVariable("key") String key) { String msg = "예제 1) XADD -> "; System.out.println(msg); // String xadd(K key, Object... keysAndValues); // String xadd(K key, Map<K, V> body); // String xadd(K key, XAddArgs args, Object... keysAndValues); // String xadd(K key, XAddArgs args, Map<K, V> body); String id; // 온도 입력 id = redisCommands.xadd(key,"temperature", "98.7"); System.out.println(id); // 온도, 습도 입력 id = redisCommands.xadd(key,"temperature", "97.7", "humidity", "50.5"); System.out.println(id); // Map으로 온도, 습도, 압력 입력 Map<String, String> data = new HashMap<>(); data.put("temperature", "90.5"); data.put("humidity", "50.5"); data.put("pressure", "10.5"); id = redisCommands.xadd(key,data); System.out.println(id); return msg+id; }

XLEN

// 예제 3) XLEN: 스트림의 엔트리(ID) 개수 리턴 // XLEN key // http://localhost:8080/xlen/sensor-1234 @GetMapping("/xlen/{key}") public String xlen(@PathVariable("key") String key) { String msg = "예제 3) XLEN -> "; Long result = redisCommands.xlen(key); msg += result; System.out.println(msg); return msg; }

XRANGE

// 예제 2) XRANGE: 데이터 조회 // XRANGE key start end [COUNT count], XREVRANGE key end start [COUNT count] // http://localhost:8080/xrange/sensor-1234 @GetMapping("/xrange/{key}") public String xrange(@PathVariable("key") String key) { String msg = "예제 2) XRANGE -> 전체 조회"; // List<StreamMessage<K, V>> xrange(K key, Range<String> range); // List<StreamMessage<K, V>> xrange(K key, Range<String> range, Limit limit); // Range<String> range = Range.unbounded(); // Range<Double> range = Range.create(2d,5d); // Range<Double> range2 = Range.from(Range.Boundary.excluding(2d),Range.Boundary.excluding(5d)); System.out.println(msg); List<StreamMessage<String, String>> result; result = redisCommands.xrange(key, Range.unbounded()); for (StreamMessage<String, String> item : result) { System.out.println("ID: "+item.getId()); Set<Map.Entry<String, String>> entries = item.getBody().entrySet(); for (Map.Entry<String, String> entry: entries) { System.out.println(" "+entry.getKey()+" "+entry.getValue()); } } msg = "예제 2) XRANGE -> 범위 지정"; System.out.println(msg); result = redisCommands.xrange(key, Range.create("1704983736200","1704983789747")); for (StreamMessage<String, String> item : result) { System.out.println("ID: "+item.getId()); Set<Map.Entry<String, String>> entries = item.getBody().entrySet(); for (Map.Entry<String, String> entry: entries) { System.out.println(" "+entry.getKey()+" "+entry.getValue()); } } return msg+"OK"; }

XREAD

// 예제 4) XREAD: 데이터 읽기 // XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...] // http://localhost:8080/xread/sensor-1234:0 -> 처음부터 읽을 때는 0을 지정 // http://localhost:8080/xread/sensor-1234:1704983789750 -> 지정한 ID 이후 데이터를 읽어온다. // http://localhost:8080/xread/sensor-1234:> -> '>' 는 XREADGROUP 명령에서만 사용할 수 있다. @GetMapping("/xread/{key}") @SuppressWarnings("unchecked") public String xread(@PathVariable("key") String key) { String msg = "예제 4) XREAD -> "; // List<StreamMessage<K, V>> xread(StreamOffset<K>... streams); // List<StreamMessage<K, V>> xread(XReadArgs args, StreamOffset<K>... streams); String[] keyId = key.split(":"); // key와 ID를 지정한다. 지정된 ID 이후 데이터를 읽어온다. XReadArgs.StreamOffset<String> stream = XReadArgs.StreamOffset.from(keyId[0],keyId[1]); // 읽어올 데이터 개수를 지정한다. 지정하지 않으면 모든 데이터를 읽어온다. XReadArgs args = XReadArgs.Builder.count(1); List<StreamMessage<String, String>> result = redisCommands.xread(args, stream); msg += result.size(); System.out.println(msg); for (StreamMessage<String, String> item : result) { System.out.println("ID: "+item.getId()); Set<Map.Entry<String, String>> entries = item.getBody().entrySet(); for (Map.Entry<String, String> entry: entries) { System.out.println(" "+entry.getKey()+" "+entry.getValue()); } } return msg; }

XDEL

// 예제 5) XDEL: 데이터 삭제 // XDEL key ID [ID ...] // http://localhost:8080/xdel/sensor-1234:1704983736197 @GetMapping("/xdel/{key}") public String xdel(@PathVariable("key") String key) { String msg = "예제 5) XDEL -> "; String[] keyId = key.split(":"); // 삭제한 id 개수를 리턴한다. Long result = redisCommands.xdel(keyId[0], keyId[1]); msg += result; System.out.println(msg); return msg; }