[RxJava] 변환 연산자 정리

[2021-04-16 업데이트]

 

[2020.12.12 블로그 포스팅 스터디 2 번째 글]

 

[변환 연산자] : 변환 연산자는 만들어진 데이터 흐름을 원하는 대로 변형할 수 있습니다. 대표적인 예로 map()과 flatMap()이 있는데 둘의 차이점은 Observable하게 만드냐 안만드냐 차이가 있습니다.  이번 포스팅에서는 flatMap()과 같은 계열인 함수들에 대해 살펴보려고 합니다.

 

[+]

flayMap()은 map()함수를 좀 더 발전시킨 함수로 map() 함수는 원하는 입력값을 어떤 함수에 넣어서 변환할 수 있는 일대일 함수이고 flatMap() 함수는 똑같이 함수에 넣더라도 결과가 Observable로 나온다는 차이점이 있습니다. 또한 flatMap()은 일대다 혹은 일대일 Observable함수입니다.

+) A FlatMap divides an Observable into many singular Observables. Hence, instead of sending each of the Observable values one by one, a FlatMap does everything in parallel and then merges the results in whatever order they are completed.

 

1. concatMap()

-> flatMap()과 비슷한 함수입니다. flatMap()는 먼저 들어온 데이터를 처리하는 도중에 새로운 데이터가 들어오면 나중에 들어온 데이터의 처리 결과가 먼저 출력될 수 있습니다. 이런것을 인터리빙이라고 합니다. 

하지만 concatMap()은 이 인터리빙을 막아주고 먼저 들어온 데이터 순서대로 처리해서 결과를 낼 수 있도록 보장해줍니다.(순서는 보장해주지만 인터리빙을 허용하는 flatMap()의 속도가 일반적으로 더 빠릅니다.) 추가로 flatMap()은 내부적으로 map() 연산자를 사용하고 concatMap()은 concat()을 사용해서 이러한 차이점이 있고 그래서 이러한 특징을 갖습니다. 

 

참고로 concat() 함수는 입력으로 들어온 여러 Observable을 연결해 인터리빙없이 순서대로 나열하는 함수입니다.

https://fernandocejas.com/2015/01/11/rxjava-observable-tranformation-concatmap-vs-flatmap/

 

 

https://fernandocejas.com/2015/01/11/rxjava-observable-tranformation-concatmap-vs-flatmap/

 

races : {"A", "B", "C"}
Observable.from(race)
                .concatMap((Func1<String, Observable<?>>) s -> {
                    final int delay = new Random().nextInt(4);
                    return Observable.just(s).map(String::toUpperCase)
                            .delay(delay, TimeUnit.SECONDS);
                })
                .subscribe(System.out::println);

        Thread.sleep(5000);

//Prints
//A
//B
//C

들어온 데이터를 처리하는 delay 처리속도가 다름에도 불구하고 순서를 보장!

 

flatMap() VS concatMap() 비교를 다음 사이트에서 잘 설명해놨으니 참고하면 됩니다.

fernandocejas.com/2015/01/11/rxjava-observable-tranformation-concatmap-vs-flatmap/

RxJava Observable tranformation: concatMap() vs flatMap() | Fernando Cejas

 

---

 

 

2. switchMap()

-> concatMap() 함수가 인터리빙이 발생할 수 있는 상황에서 순서를 보장해준다면 switchMap() 함수는 순서를 보장하기 위해 기존에 진행 중이던 작업을 바로 중단합니다.  그래서 여러 개의 값이 발행되었을때 마지막에 들어온 값만 처리하고 싶을 때 사용합니다. (중간에 끊키더라도 마지막 데이터 처리는 보장되기 때문) 이러한 이유로  대표적으로 최종값만 필요한 센서 등의 값을 얻어와서 동적으로 처리하는 경우에 많이 유용하다고 합니다.

 

초록을 처리하다가 중간에 파랑이 들어와서 초록의 처리가 중단되고 파랑을 처리합니다.

@Test
public void switchMap() throws Exception {
    final List<String> items = Lists.newArrayList("a", "b", "c", "d", "e", "f");

    final TestScheduler scheduler = new TestScheduler();

    Observable.from(items)
            .switchMap( s -> {
                final int delay = new Random().nextInt(10);
                return Observable.just(s + "x")
                        .delay(delay, TimeUnit.SECONDS, scheduler);
            })
            .toList()
            .doOnNext(System.out::println)
            .subscribe();

    scheduler.advanceTimeBy(1, TimeUnit.MINUTES);
}
//마지막 처리값인 fx만 출력이 된다.
//출처 : https://medium.com/appunite-edu-collection/rxjava-flatmap-switchmap-and-concatmap-differences-examples-6d1f3ff88ee0

 

 

 

---

 

 

3. groupBy()

-> 이 함수는 어떤 기준(keySelector 인자)으로 단일 Observable을 여러 개로 이루어진 Observable 그룹(GroupedObservable)으로 만듭니다.

GroupedObservable은 Observable과 동일하지만 getKey() 라는 메서드를 제공하여 구분된 그룹을 알 수 있게 해줍니다.

 

[공식문서]

The GroupBy operator divides an Observable that emits items into an Observable that emits Observables, each one of which emits some subset of the items from the original source Observable. Which items end up on which Observable is typically decided by a discriminating function that evaluates each item and assigns it a key. All items with the same key are emitted by the same Observable.

 

다음 좋은 예제가 있어서 갖고와봤습니다.

Artice 객체 배열을 RFID_TAG_ID로 그룹화 하고 그 그룹화 된 것을 map으로 하나씩 가져와서 ArticleQuantity 객체를 생성하고 reduce()를 통해 합쳐  RFID_TAG_ID와 해당 ID의 값(개수)를 담은 ArticleQuantity로 최종결과를 합성해줍니다. 

[+]

reduce() 함수는 발행한 데이터를 모두 사용하여 어떤 최종 결과 데이터를 합성할 때 활용하는 함수입니다.(그룹 내의 모든 항목을 더하기 위해 reduce라는 오퍼레이터를 사용합니다. 이 오퍼레이터는 두 개의 객체를 받아서 한 개의 객체를 생성합니다.)

 

출처 : academy.realm.io/kr/posts/mobilization-hugo-visser-rxjava-for-rest-of-us/

누구나 할 수 있는 RxJava, 지금 시작하기!

 

 

 

 

 

4. scan()

->reduce()와 비슷한 함수입니다. reduce() 함수는 Observable에서 모든 데이터가 입력된 후 그것을 종합하여 마지막 1개의 데이터만을 구독자에게 발행했습니다. 반면 scan() 함수는 실행할 때마다 입력값에 맞는 중간 결과 및 최종 결과를 구독자에게 발행합니다. 

 

둘의 차이점에 대해 추가로 설명하면 reduce()는 마지막 값이 입력되지 않거나 onComplete 이벤트가 발생하지 않으면 구독자에게 발행하지 않으므로 최악의 경우에는 값을 전혀 발생하지 않고 종료될 수도 있으므로 Maybe 리턴 클래스 타입을 갖습니다. scan()은 Observable 타입을 갖습니다.

 

 

 

[+] taeiim.tistory.com/entry/RxJava2-2-Observable-Single-Maybe-%EB%9C%A8%EA%B1%B0%EC%9A%B4%EC%B0%A8%EA%B0%80%EC%9A%B4-Observable-%ED%8C%A9%ED%86%A0%EB%A6%AC%ED%95%A8%EC%88%98

Maybe : Single 클래스와 마찬가지로 최대 데이터 하나를 가질 수 있지만 데이터 발행 없이 바로 데이터 발생을 완료할 수 있다.

• Single : 1개 완료 / Maybe : 0 or 1 완료

들어올때마다 이전의 값들과 합성하여 방출하는 것을 볼 수 있습니다.

Observable.just(1, 2, 3, 4, 5)
    .scan(new Func2<Integer, Integer, Integer>() {
        @Override
        public Integer call(Integer sum, Integer item) {
            return sum + item;
        }
    }).subscribe(new Subscriber<Integer>() {
        @Override
        public void onNext(Integer item) {
            System.out.println("Next: " + item);
        }

        @Override
        public void onError(Throwable error) {
            System.err.println("Error: " + error.getMessage());
        }

        @Override
        public void onCompleted() {
            System.out.println("Sequence complete.");
        }
    });
    
    //결과
    Next: 1
	Next: 3
	Next: 6
	Next: 10
	Next: 15
	Sequence complete.

 

 

---

이외 참고 :

RxJava 프로그래밍

reactivex.io/documentation/operators.html

ReactiveX - Operators

github.com/yudong80/reactivejava/tree/master/src/main/java/com/yudong80/reactivejava/chapter04/create

yudong80/reactivejava

 

 

 

 

 

 

 

댓글과 공감은 큰 힘이 됩니다. 감사합니다. !!