[RxJava] Operators - 변환

안녕하세요 점냥입니다 :)

 

지난 포스팅에서 Rx Operator의 주요 생성 함수들을 알아보았고 이번 포스팅에서는 변환해주는 Rx Operator에 대해서 알아보겠습니다.

만약 다른 RxJava의 주요 함수에 대해서 알고 싶다면 공식 링크를 클릭해주세요

 

RxJava Operators 분류

생성	just(), fromArray(), fromIterator(), create(), Interval()
변환	map(), flatMap(), cancatMap(), switchMap(), reduce(), scan()
제어	filter(), take(), skip()
결합	zip(), combineLatest()

 

 

map - 기본적인 데이터 변환

        String[] nums = new String[] {"1", "2", "3"};
        
        Observable.fromArray(nums)
                .map(Integer::parseInt)
                .subscribe(System.out::println);
        
        출력
        >> 1
        >> 2
        >> 3

map은 데이터를 데이터로 변환시키는 함수입니다. 이게 무슨말이냐 함은 데이터 한개를 변환하여 데이터 한개가 나온다라는 뜻입니다.

 

코드를 보면 nums 변수는 String 배열 객체이고, 이 변수를 이용해 데이터 흐름을 생성하기 때문에 초기 데이터들은 String 타입의 데이터 들입니다. 그런데 map 함수로 인해 String이 Integer 타입의 데이터로 변환되게 됩니다

 

 

flatMap - 데이터를 또 다른 데이터 흐름으로 변환

        Observable.interval(100L, TimeUnit.MILLISECONDS)
                .take(3)
                .flatMap(data -> Observable.interval(200L, TimeUnit.MILLISECONDS)
                		.take(2)
                		.map(val -> "data: " + data + " value: " + val))
                .subscribe(System.out::println);
        
        출력
        >> data: 0 value: 0
        >> data: 1 value: 0
        >> data: 2 value: 0
        >> data: 0 value: 1
        >> data: 1 value: 1
        >> data: 2 value: 1

flatMap은 단일 데이터를 또 다른 데이터 흐름으로 변환하는 함수입니다. 여기서 또 다른 데이터 흐름이란 Observable를 의미해요!

 

Interval 생성함수는 각각의 데이터 발행마다 매개변수의 시간만큼 지연되고 데이터는 0부터 1씩 증가된 데이터들이 무한히 발행되죠. 

하지만 나중에 배울 take(3)에 의해서 3번 발행되는 것으로 변경됩니다.

 

따라서 데이터는 [0, 1, 2]가 생성이 되고 이 각 데이터들은 flatMap함수에서 또 Interval 함수로 인해 지연시간을 가지며 take(2) 함수로 인해 2개의 데이터 흐름으로 변환이 됩니다. 그리고 그 새로운 데이터 흐름의 데이터들은 최종적으로 map 함수로 인해 출력값에 나타나는 String 데이터로 변환이 됩니다. 많이 복잡하네요 ㅠㅠ

 

그런데 출력 결과를 보면 입력 순서와 비교해서 동일하지 않다는 것을 알수있어요!

 

 

 

 

 

ReactiveX - FlatMap

 

위의 의문점을 FlatMap 마블 다이어그램으로 좀 더 이해해봅시다. 

오른쪽 방향을 가리키는 화살표들은 데이터 흐름을 의미하고 그 안의 도형은 데이터를 의미해요. 맨 위에 있는 데이터 흐름에서 3가지 색상의 데이터가 흐르는 것을 확인할 수 있습니다.

 

중간의 네모 도형을 보면 FlatMap의 함수 기능을 나타낸 다는 것을 알 수 있고, 하나의 원 도형을 2개의 마름모 도형을 생성하여 아래 데이터 흐름에서 원 도형이 2개의 마름모로 변환되는 것을 볼 수 있죠! 그런데 특이한 점이 있습니다.

 

초기 데이터 흐름에서 초록 원 도형과 파란 원 도형이 순차적으로 FlatMap을 통해 2개의 마름모로 변환되는 데 새로 변환된 데이터 흐름을 보면 2개의 초록 마름모와 사이에 파란 마름모가 끼어 있는 것을 볼 수 있어요. 이 그림을 통해 알 수 있는 FlatMap은 변환된 데이터 흐름은 입력된 데이터 순서대로 새로운 데이터 흐름이 발행되는 것을 보장하지 않는다는 것이에요.

 

 

concatMap - 순서가 보장되는 FlatMap

        Observable.interval(100L, TimeUnit.MILLISECONDS)
                .take(3)
                .concatMap(data -> Observable.interval(200L, TimeUnit.MILLISECONDS)
                		.take(2)
                		.map(val -> "data: " + data + " value: " + val))
                .subscribe(System.out::println);
                
                
        출력
        >> data: 0 value: 0
        >> data: 0 value: 1
        >> data: 1 value: 0
        >> data: 1 value: 1
        >> data: 2 value: 0
        >> data: 2 value: 1

 

ReactiveX - concatMap

concatMap은 flatMap과 사용법이 완전 똑같아요! 다른점은 flatMap과 달리 입력 데이터의 순서를 보장해준 다는 것입니다.

대신 입력 순서를 보장해주는 만큼 데이터 간 기다리는 지연 시간이 생기기 때문에 총 실행 시간이 길어져요.

 

 

switchMap - 가장 최신 데이터를 보장하며 데이터 흐름을 생성

        Observable.interval(100L, TimeUnit.MILLISECONDS)
                .take(3)
                .switchMap(data -> Observable.interval(200L, TimeUnit.MILLISECONDS)
                		.take(2)
                		.map(val -> "data: " + data + " value: " + val))
                .subscribe(System.out::println);
                
        출력
        >> data: 2 value: 0
        >> data: 2 value: 1

switchMap의 코드를 보면 flatMap과 concatMap처럼 함수 호출만 변경된 것을 볼 수 있어요.

그런데 결과를 보면 Interval().take(3)의 조합으로 [0, 1, 2] 데이터 흐름이 생성이 되는데 출력을 보면 제일 마지막 2의 데이터로 인해 변환된 새로운 데이터 흐름의 결과 값만 출력된 것을 볼 수 있습니다.

 

마치 switchMap이 가장 마지막 데이터 흐름에 대해서만 변환하는것 같은데요 아닙니다!

 

ReactiveX - switchMap

 

SwitchMap은 flatmap과 concatMap과 동일하게 모든 데이터 입력이 들어오면 새로운 데이터 흐름으로 변환하려고 시도합니다. 그런데

변환 작업 중 새로운 데이터 입력이 들어오게 되면 기존 작업을 취소하고 새로운 데이터에 대해서 변환 작업을 진행합니다.

 

코드을 보면 데이터의 입력 지연시간이 각 100ms, 변환 시간이 200ms이기 때문에 데이터 변환 시간보다 입력 시간이 훨씬 빠릅니다.

따라서 변환 도중 새로운 입력이 들어오게 되어 기존 작업을 취소 하게 되고 최종적으로 마지막 입력 데이터에 대해서 변환을 진행하게 되었습니다. 이러한 switchMap의 특성은 실시간으로 최신 데이터를 가져올 때 유용하게 쓰이고 있어요 :)

 

 

reduce - 모든 데이터를 조합(변환)하여 최종 값 출력

        String[] balls = new String[] {"A", "B", "C"};

        Observable.fromArray(balls)
                .reduce((ball1, ball2) -> ball2 + "(" + ball1 + ")")
                .subscribe(System.out::println);
                
        출력
        >> C(B(A))

 

reduce 함수는 데이터를 중첩되게 변환시켜 최종적인 새로운 변환된 단일 데이터를 만듭니다. 

 

ReactiveX - reduce

 

추가적으로 마블 다이어 그램을 보면 더 이해하기 쉬우실 거에요. 데이터로 변환 된 값이 구독자에게 전달되는 것이 아닌 다시 reduce 함수의 입력 값으로 전달되어 새로운 변환된 데이터를 생성하고 다시 입력값으로 들어가는 이러한 과정이 데이터 흐름의 onComplete() 이벤트가 발생할 때까지 진행됩니다.

 

 

 

scan() - 모든 데이터 조합(변환)하며 각 결과 값 출력

        String[] balls = new String[] {"A", "B", "C"};

        Observable.fromArray(balls)
                .scan((ball1, ball2) -> ball2 + "(" + ball1 + ")")
                .subscribe(System.out::println);
                
        출력
        >> A
        >> B(A)
        >> C(B(A))

 

scan은 reduce() 함수가 각 변환 된 값이 모여 최종 값을 출력한 것이라면 scan은 각 변환 된 중간 결과 값들을 모두 출력하는 것으로 이해하면 됩니다!