[RxJava2] 리액티브 변환 연산자 : concatMap(), switchMap(), groupBy(), scan()

공감 및 댓글은 포스팅 하는데  아주아주 큰 힘이 됩니다!! 포스팅 내용이 찾아주신 분들께  도움이 되길 바라며 더 깔끔하고 좋은 포스팅을  만들어 나가겠습니다^^

이번 포스팅에서는 변환 연산자 4 개에 대해 공부합니다.

변환 연산자

만들어진 데이터의 흐름을 원하는 대로 변형할 수 있다.

concatMap()

이전에 배운 flatMap()은 먼저 들어온 데이터를 처리하는 도중 새로운 데이터가

들어오게 되면 나중에 들어온 데이터가 먼저 출력될 수도 있는데,

이것을 인터리빙이라고 합니다.

concatMap()은 이 인터리빙 없습니다.

즉. 항상 먼저 들어온 데이터를 먼저 처리합니다. => 순서 보장.

예제와 주석.

package rx.java.chapter04.transform;
 
import java.util.concurrent.TimeUnit;
 
import io.reactivex.Observable;
 
public class ConcatMapEx {
 
    public static void main(String[] args) {
 
        String[] names = {"AAA", "BBB", "CCC"};
        
        Observable<String> src = Observable.interval(100L, TimeUnit.MILLISECONDS)
                .map(idx -> names[idx.intValue()])
                .take(names.length)
                .concatMap(name -> Observable.interval(200L, TimeUnit.MILLISECONDS)
                        .map(unUsed -> "*" + name + "*")
                        .take(2)
                    );
        src.subscribe(System.out::println);
        try {
            Thread.sleep(2000);
        } catch (Exception e) {
            // TODO: handle exception
        }
        
        System.out.println("====================");
        
        /*
         *
         * flatMap() 과 concatMap()의 차이는 인터리빙의 허용 유무
         * 
         * 그 외의 사용법은 동일하다. 순서가 중요한 곳에서는 꼭 concatMap()을 사용하도록.
         * 
         * But flatMap() 과 concatMap() 중 flatMap()이 더 빠름, 순서가 중요치 않으므로
         * 
         */
        
        Observable<String> src2 = Observable.interval(100L, TimeUnit.MILLISECONDS)
                .map(idx -> names[idx.intValue()])
                .take(names.length)
                .flatMap(name -> Observable.interval(200L, TimeUnit.MILLISECONDS)
                        .map(unUsed -> "*" + name + "*")
                        .take(2)
                    );
        src2.subscribe(System.out::println);
        try {
            Thread.sleep(2000);
        } catch (Exception e) {
            // TODO: handle exception
        }
    }
 
}
 

concatMap()은 순서가 보장되는 대신 속도가 조금 느리고,

flatMap()은 빠른 대신 순서가 보장되지 않아서 매번 실행 결과가 다르게 나옵니다.

switchMap()

concatMap() 처럼 함수의 순서를 보장해주는 역할은 같지만,

switchMap()은 순서 보장을 위해 기존에 진행 중이던 작업을 바로 중단합니다.

여러 개의 값이 발행되었을 때 마지막에 들어온 값만 처리하고 싶을 때 사용합니다.

package rx.java.chapter04.transform;
 
import java.util.concurrent.TimeUnit;
 
import io.reactivex.Observable;
 
public class SwitchMapEx {
 
    public static void main(String[] args) {
 
        String[] names = {"AAA", "BBB", "CCC", "DDD", "EEE"};
        
        Observable<String> src = Observable.interval(100L, TimeUnit.MILLISECONDS)
                .map(idx -> names[idx.intValue()])
                .take(names.length)
                .switchMap(name -> Observable.interval(200L, TimeUnit.MILLISECONDS)
                        .map(unUsed -> Thread.currentThread().getName() + " => *" + name + "*")
                        .take(2)
                    );
        src.subscribe(System.out::println);
        try {
            Thread.sleep(2000);
        } catch (Exception e) {
            // TODO: handle exception
        }
        
        
    }
 
}
 

groupBy()

단일 Observable 을 여러 개로 이루어진 Observable 그룹으로 만듭니다.

package rx.java.chapter04.transform;
 
import io.reactivex.Observable;
import io.reactivex.functions.Function;
import io.reactivex.observables.GroupedObservable;
 
public class GroupByEx {
 
    public static void main(String[] args) {
 
        String[] groups = {"AAA", "AAA-T", "BBB", "CCC", "BBB-T", "CCC-T"};
        
        Function<String, String> func = group -> {
            if(group.endsWith("-T")) return "T Group";
            return "Normal Group";
        };
        
        
        Observable<GroupedObservable<String, String>> src = 
                Observable.fromArray(groups).groupBy(func);
        
        //2 개의 그룹이 생겼음을 알 수 있다.
        src.subscribe(System.out::println);
        
        src.subscribe(srcs -> {
            srcs.subscribe(val -> System.out.println("GROUP : " + srcs.getKey() + "\t Value : " + val));
        });
        
    }
 
}
 

scan()

reduce() 함수와 비슷합니다.

차이점은 reduce() 함수는 Observable에서 모든 데이터가 입력된 후에 데이터를 

종합해 마지막 1개의 데이터만 구독자에게 발행했지만,

scan() 함수는 실행할 때마다 입력값에 맞는 중간 값과 최종 값을 구독자에게

발행합니다.

package rx.java.chapter04.transform;
 
import io.reactivex.Observable;
 
public class ScanEx {
 
    public static void main(String[] args) {
 
        //reduce() 함수와 비교
        
        String[] names = {"Park", "Thomas", "Sunggyun"};
        
        Observable<String> src = Observable.fromArray(names)
                .scan((val1, val2) -> val2 + " " + val1);
        
        src.subscribe(System.out::println);
                
        
    }
 
}
 

다음 포스팅에서는 결합 연산자에 대해 공부합니다.