[Rxjava] 변환 연산자(map, flatMap, groupBy)

이번 포스팅에선 변환 연산자 중 map, flatMap, groupBy에 대해 알아본다.

 

map

 

ReactiveX 공식 홈페이지에선 map 연산자를 아래와 같이 설명하고 있다.

http://reactivex.io/documentation/operators/map.html

ReactiveX - Map operator

각 아이템에 함수를 적용해 Observable에서 방출하는 항목을 변환한다.
map 연산자는 소스 Observable이 내보낸 각 아이템에 선택한 함수를 적용하고 이런 함수 응용 프로그램의 결과를 내보내는 Observable을 반환한다

 

정리하면 Observable로 어떤 아이템들을 방출시킬 때 내가 선택한 함수를 적용하고 그 결과값을 리턴한다는 뜻이다.

간단한 예시로 확인해보자.

import io.reactivex.rxjava3.core.Observable;

public class Test {
    public static void main(String[] args) {
        String[] numbers = {"1","2","3","4"};
        Observable<String> source = Observable.fromArray(numbers)
                .map(num -> num + " / 여기가 추가됩니다");
        source.subscribe(System.out::println);
    }
}

또는 이렇게도 쓸 수 있다.

public class Test {
    public static void main(String[] args) {
        String[] numbers = {"1","2","3","4"};
        Observable.fromArray(numbers)
                .flatMap(data -> Observable.just(data).map(result -> data + "*" + data))
                .subscribe(justResult -> System.out.println("결과 : " + justResult));
    }
}

flatMap()이라는 생소한 연산자가 보이지만 코드를 보면 flatMap() 안에서 just()와 map()을 호출해 데이터를 변형시키는 게 보인다.

그 다음 subscribe() 안에서 justResult라는 변수를 통해 변형된 데이터를 출력한다. String[] 안에 4개의 데이터가 들어있으니 당연히 4번 수행되면 Observable이 종료된다.

 

 

flatMap

 

그럼 flatMap은 뭘까? 위에서  보았듯 flatMap() 안에 2가지 연산자를 사용해 데이터를 변형시킬 수 있다.

이걸 생각하면서 아래의 공식 문서를 읽어보자.

http://reactivex.io/documentation/operators/flatmap.html

ReactiveX - FlatMap operator

Observable에 의해 방출된 아이템을 Observable로 변환한 다음 그 방출을 단일 Observable로 평면화한다.
flatMap()은 소스 Observable이 내보낸 각 아이템에 지정한 함수를 적용해 Observable을 변환한다. 여기서 해당 함수는 자체적으로 아이템을 내보내는 Observable을 반환한다. 그 다음 flatMap은 결과 Observable의 방출을 병합해서 그 결과를 자체 시퀀스로 방출한다.
이 방법은 예를 들어 자체적으로 Observable 멤버가 있거나 다른 방식으로 Observable로 변환 가능한 일련의 아이템을 방출하는 Observable이 있을 때 유용하다. 이런 항목의 하위 Observable이다. flatMap은 이런 Observable의 방출을 병합해 interleave할 수 있다. 여러 언어 별 구현에는 변환된 Observable의 방출을 interleave하지 않고, 대신 이런 방출을 엄격한 순서로 방출하는 연산자가 있으며, 종종 concatMap 또는 이와 유사하게 불린다

flatMap에 의해 소스 Observable의 아이템에 매핑된 개별 Observable이 onError()를 호출해서 중단되면 flatMap에 의해 생성된 Observable이 즉시 중단되고 onError()를 호출한다. 이 연산자의 변형 버전은 추가 int 매개변수를 사용한다. 이 매개변수는 flatMap이 원본 Observable에서 내보낸 아이템이 매핑하는 Observable에 대해 시도할 최대 동시 구독 수를 설정한다. 이 최대 수에 도달하면 다른 Observable을 구독하기 전에 해당 Observable 중 하나가 종료될 때까지 기다린다.

 

정리하면 Observable이 방출한 아이템 별로 내가 원하는 함수를 적용할 수 있게 해주는 연산자란 뜻 같다. 리턴값은 함수가 적용된 결과의 시퀀스다.

예전에 Rxjava로 구구단을 출력하는 예제를 포스팅했었는데, 그 포스팅의 코드에 flatMap()이 섞여 있다. 아래 코드를 참고하자.

 

@Override
public void onTextChanged(CharSequence s, int start, int before, int count)
{
    subject.map(dan -> editText.getText().toString().equals(""))
            .flatMap(dan -> BehaviorSubject.range(1, 9),
                    (dan, row) -> 0 + "x" + row + " = " + 0 + "\n")
            .scan((x, y) -> x + y)
            .doOnNext(data -> Log.d("1번째 doOnNext()", data))
            .subscribe(text -> textView.setText(text));

    subject.map(dan -> Long.parseLong(editText.getText().toString()))
            .flatMap(dan -> BehaviorSubject.range(1, 9),
                    (dan, row) -> dan + " x " + row + " = " + (dan * row) + "\n")
            .scan((x, y) -> x + y)
            .doOnNext(data -> Log.d("2번째 doOnNext()", data))
            .subscribe(text -> textView.setText(text), Throwable::getMessage);
}

 

첫 번째 map() 안에서 1~9까지 방출시킨 다음 (dan, row) 형태로 2개의 변수를 새로 정의하고 텍스트뷰에 붙일 문자열을 변형해 setText()로 텍스트뷰에 붙이고 있다.

두 번째 map() 안에서도 이와 비슷한 처리가 이뤄지고 있는 게 보인다. 차이라면 두 번째 map() 안에선 실제 곱셈 처리가 수행된다는 것, 에러 발생 시 어떤 에러인지 메시지를 던지는 처리가 있단 것 정도다.

안드로이드의 경우 map, flatMap 연산자를 잘 활용하면 여러 API에서 값을 가져온 다음 이걸 합치는 등 가공 처리해서 뷰에 붙이거나 저장하는 등 다양한 처리가 가능할 것 같다.

 

groupBy

 

이름부터 어떤 기준으로 뭔가를 그룹 지을 것 같다는 느낌이 든다. 이 뇌피셜이 맞는지 공식 문서로 확인해보자.

 

http://reactivex.io/documentation/operators/groupby.html

ReactiveX - GroupBy operator

Observable을 원래 Observable과 다른 아이템 하위 집합을 내보내는 Observable 집합으로 나눈다.
groupBy()는 아이템을 내보내는 Observable을 Observable을 내보내는 Observable로 나눈다. 각 Observable은 원래 소스 Observable에서 아이템의 일부 하위 집합을 방출한다. Observable로 끝나는 아이템은 일반적으로 각 아이템을 평가하고 키를 할당하는 식별 기능에 의해 결정된다. 동일한 키를 가진 모든 아이템은 동일한 Observable에 의해 방출된다

이것이 반환하는 Observable은 Observable의 특정 하위 클래스인 GroupedObservable의 아이템을 방출한다. GroupedObservable 인터페이스를 구현하는 객체에는 이 특정 GroupedObservable에 대해 지정된 항목의 키를 검색할 수 있는 추가 메서드인 getKey()가 있다. groupBy()의 다른 버전을 사용하면 결과 GroupedObservable에서 아이템을 방출하기 전에 요소를 변경하는 변형 함수를 전달할 수 있다.
groupBy()가 소스 Observable을 GroupedObservable을 방출하는 Observable로 분할할 때 이러한 GroupedObservable 각각은 구독 시 방출할 아이템을 버퍼링하기 시작한다. 이런 이유로 GroupedObservables 중 하나를 무시하면(구독하거나 구독하는 연산자를 적용하지 않음) 이 버퍼는 잠재적으로 메모리 누수를 일으킨다...(중략)...groupBy()는 기본적으로 특정 스케줄러에서 작동하지 않는다.

 

정리하면 groupBy()는 하나의 Observable을 여러 Observable 그룹(GroupedObservable, 그룹지어진 Observable)로 만드는 연산자다. 물론 아무렇게나 나누는 게 아니라 특정 기준이 있어서 이를 기준으로 나누는 것이다.

아래는 groupBy()를 사용하는 간단한 예시다. 먼저 아래의 Shape 클래스를 만든다.

 

public class Shape {
    public static String getShape(String obj) {
        if (obj == null || obj.equals("")) return "NO_SHAPE";
        if (obj.endsWith("-H")) return "HEXAGON";
        if (obj.endsWith("-O")) return "OCTAGON";
        if (obj.endsWith("-R")) return "RECTANGLE";
        if (obj.endsWith("-T")) return "TRIANGLE";
        if (obj.endsWith("<>")) return "DIAMOND";
        if (obj.endsWith("-P")) return "PENTAGON";
        if (obj.endsWith("-S")) return "STAR";
        return "BALL";
    }
}

 

그리고 이 클래스를 메인 메서드에서 가져와 사용한다.

 

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.observables.GroupedObservable;

public class Test {
    public static void main(String[] args) {
        String[] elements = {"6", "4", "2-T", "2", "6-T", "4-T"};
        Observable<GroupedObservable<String, String>> source =Observable.fromArray(elements)
                .groupBy(Shape::getShape);
        source.subscribe(result -> {
            result.subscribe(value -> {
                System.out.println("그룹 : " + result.getKey() + "\t 값 : " + value);
            });
        });
    }
}

 

참고로 한 번에 나오기 위해서 중괄호를 여러 번 둘러쳤는데 굳이 저게 없어도 한 줄로 작성 가능하다.

결과를 보면 "-T"가 붙은 아이템은 TRIANGLE로 분류되고, 그 외는 전부 BALL로 분류되는 걸 볼 수 있다.

 

 

참고한 사이트)

 

https://beomseok95.tistory.com/33

리액티브연산자[변환 연산자]-13(groupBy, scan)

 

https://gamjatwigim.tistory.com/51

RxJava 공부 3 - Map과 FlatMap, Timer와 interval, FlatMap과 ConcatMap