Lettuce Reactive

Redis Developer Course	Redis Technical Support	Redis Enterprise Server

Lettuce Reactive

Lettuce Reactive로 구현한 Redis 대표 명령들

Strings 대표 명령들

// SET
protected void set() {
	long idx2 = idx++;
	reactiveCmd.set(key+idx2, value+idx2)
		 .subscribe(v -> System.out.println("SET "+key+idx2+" "+value+idx2+" -> "+v));
}

// GET
protected void get() {
	long idx2 = random.nextInt(10);
	reactiveCmd.get(key+idx2)
		 .subscribe(v -> System.out.println("GET "+key+idx2+" -> "+v));
}

// DEL
protected void del() {
	long idx2 = random.nextInt(10);
	reactiveCmd.del(key+idx2)
		 .subscribe(v -> System.out.println("DEL "+key+idx2+" -> "+v));
}

// INCR
protected void incr() {
	reactiveCmd.incr("page")
		 .subscribe(v -> System.out.println("INCR page -> "+v));
}

Lists 대표 명령들

// LPUSH
protected void lpush() {
	idx++;
	reactiveCmd.lpush(mylist, "ele-"+idx)
		.subscribe(v -> System.out.println("LPUSH "+mylist+" ele"+idx+" -> "+v));
}

// RPOP
protected void rpop() {
	reactiveCmd.rpop(mylist)
		.subscribe(v -> System.out.println("RPOP "+mylist+" -> "+v));
}

// LRANGE
protected void lrange() {
	reactiveCmd.lrange(mylist, 0, -1)
		.subscribe(v -> System.out.println(v));
}

Sets 대표 명령들

// SADD
protected void sadd() {
	idx++;
	reactiveCmd.sadd(myset, "mem"+idx)
		.subscribe(v -> System.out.println("SADD "+myset+" mem"+idx+" -> "+v));
}

// SREM
protected void srem() {
	idx = random.nextInt(10);
	reactiveCmd.srem(myset, "mem"+idx)
		.subscribe(v -> System.out.println("SREM "+myset+" mem"+idx+" -> "+v));
}

// SMEMBERS
protected void smembers() {
	reactiveCmd.smembers(myset)
		.subscribe(v -> System.out.println(v));
}

ZSets 대표 명령들

// ZADD
protected void zadd() {
	idx++;
	reactiveCmd.zadd(myzset, idx, "mem"+idx)
		.subscribe(v -> System.out.println("ZADD "+myzset+" mem"+idx+" -> "+v));
}

// ZREM
protected void zrem() {
	idx = random.nextInt(10);
	reactiveCmd.zrem(myzset, "mem"+idx)
		.subscribe(v -> System.out.println("ZREM "+myzset+" mem"+idx+" -> "+v));
}

// ZRANGE
protected void zrange() {
	System.out.println("zrange");
	reactiveCmd.zrange(myzset, 0, -1)
		.subscribe(v -> System.out.println(v));
	System.out.println("zrangeWithScore");
	reactiveCmd.zrangeWithScores(myzset, 0, -1)
		.subscribe(v -> System.out.println(v));
}

Hashes 대표 명령들

// HSET
protected void hset() {
	idx++;
	reactiveCmd.hset(myhash, "field"+idx, "value"+idx)
		.subscribe(v -> System.out.println("HSET "+"field"+idx+" value"+idx+" -> "+v));
}

// HDEL
protected void hdel() {
	idx = random.nextInt(10);
	reactiveCmd.hdel(myhash, "field"+idx)
		.subscribe(v -> System.out.println("HDEL field"+idx+" -> "+v));
}

// HGETALL
protected void hgetall() {
	reactiveCmd.hgetall(myhash)
		.subscribe(v -> v.entrySet().forEach(System.out::println));
}

성능 테스트

protected void setSync() {
	int loop = 100000;
	System.out.println("SET Sync Test "+loop+" times --------------");
	long startTime = System.currentTimeMillis();
	System.out.println("Start time: "+new Timestamp(startTime));
	for (int i=0; i<loop; i++) {
		Redis.syncCmd.set("keyReactiveSync"+i, "valueReactiveSync"+i);
	}
	System.out.println("SET Sync elapsed time(ms) -> "+(System.currentTimeMillis() - startTime)); 
}

protected void setReactive() {
	int loop = 100000;
	System.out.println("SET Reactive Async Test "+loop+" times --------------");
	long startTime = System.currentTimeMillis();
	System.out.println("Start time: "+new Timestamp(startTime));
	for (int i=0; i<loop; i++) {
		reactiveCmd.set("keyReactiveAsync"+i, "valueReactiveAsync"+i)
			.subscribe();
	}
	System.out.println("SET Reactive elapsed time(ms) -> "+(System.currentTimeMillis() - startTime)); 
}

에러 처리

// Error handling
protected void errorHandling1() {
	Observable.just("Ben", "Michael", "Mark").doOnNext(new Action1<String>() {
			@Override
			public void call(String s) {
					throw new IllegalStateException("Takes way too long");
			}
	}).onErrorReturn(new Func1<Throwable, String>() {
			@Override
			public String call(Throwable throwable) {
					return "Default value";
			}
	}).subscribe(v -> System.out.println(v));
}

protected void errorHandling2() {
	Observable.just("Ben", "Michael", "Mark").doOnNext(new Action1<String>() {
			@Override
			public void call(String s) {
					throw new IllegalStateException("Takes way too long");
			}
	}).onErrorResumeNext(reactiveCmd.get(key+1))
	.subscribe(v -> System.out.println(v));
}

protected void errorHandling3() {
	Observable.just("Ben", "Michael", "Mark").flatMap(new Func1<String, Observable<String>>() {
			@Override
			public Observable<String> call(String key) {
					return reactiveCmd.get(key);
			}
	}).retry().subscribe(v -> System.out.println(v));
}

protected void errorHandling4() {
	Observable.just("Ben", "Michael", "Mark").flatMap(new Func1<String, Observable<?>>() {
			@Override
			public Observable<?> call(String key) {
					return reactiveCmd.get(key);
			}
	}).retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
			@Override
			public Observable<?> call(Observable<? extends Throwable> attempts) {

					return attempts.zipWith(Observable.range(1, 3), new Func2<Throwable, Integer, Integer>() {
							@Override
							public Integer call(Throwable throwable, Integer integer) {
									return integer;
							}
					}).flatMap(new Func1<Integer, Observable<Long>>() {
							@Override
							public Observable<Long> call(Integer i) {
									{
											System.out.println("delay retry by " + i + " second(s)");
											return Observable.timer(i, TimeUnit.SECONDS);
									}
							}
					});
			}
	}).subscribe(v -> System.out.println(v));
}

protected void errorHandling5() {
	Observable.just("Ben", "Michael", "Mark")
		 .flatMap(key -> reactiveCmd.get(key))
		 .retryWhen(attempts ->
				attempts
					 .zipWith(Observable.range(1, 3), (throwable, integer) -> integer)
					 .flatMap(i -> {
							System.out.println("delay retry by " + i + " second(s)");
							return Observable.timer(i, TimeUnit.SECONDS);
						}))
		 .subscribe(v -> System.out.println(v));
}

<< Async	Reactive	Connection Pooling >>

Email 답글이 올라오면 이메일로 알려드리겠습니다.