[RxJava] #3 리액티브 연산자 입문

-

ReactiveX 문서의 연산자 부분을 살펴보면 다양한 연산자(operator)함수가 존재한다.

이 연산자들은 버전이 높아지면서 계속 늘고 있다.

이 연산자들을 모두 알아야 리액티브 프로그래밍을 할 수 있는 것은 아니다.

-

for, if while 문처럼 프로그래밍 언어에서 제공하는 키워드들을 사용하는 대신 RxJava 의 연산자로 비동기 프로그래밍에 필요한 주요 로직을 만들 수 있다.

-

리액티브 연산자의 특징은 언어 특성과 크게 연관이 없다.

리액티브 연산자는 함수라고 한다.

형식만 놓고 보면 Observable 등의 클래스에 포함되어 있으므로 자바 관점에서는 메서드이다.

하지만 함수형 프로그래밍의 원리에 따르면 리엑티브 연산자는 부수 효과가 없는 순수 함수(pure function)이다. 

-

리엑티브 연산자의 분류는 아래와 같다.

생성(Creating) 연산자

    Observable, Single 클래스 등으로 데이터의 흐름을 만들어내는 함수.

    create(), just(), fromArray(), interval(), range(), timer(), defer()

변환(Transforming) 연산자

    입력을 받아서 원하는 출력 결과를 내는 전통적인 의미의 함수.

    map(), flatMap()

필터(Filter) 연산자

    입력 데이터 중에 원하는 데이터만 걸러낸다.

    filter(), first(), take()

합성(Combining) 연산자

    합성 연산자는 여러 Observable 을 조립하는 역할을 한다.

오류 처리(Error Handling) 연산자

    onErrorReturn(), onErrorResumeNext(), retry() 등이 있다.

유틸리티(Utility) 연산자

    subscribeOn(), observeOn() 등이 있으며 비동기 프로그래밍을 지원한다.

조건(Conditional) 연산자

    Observable 의 흐름을 제어하는 역할을 한다.

수학과 집합형(Mathematical and Aggregate) 연산자

    수학 함수와 연관 있는 연산자.

배압(Back pressure) 연산자

    배압 이슈에 대응하는 연산자

3.1. map() 함수

-

@CheckReturnValue
@SchedulerSupport(value=“none”)
public final <R> Observable<R> map(Function<? super T, ? extends R> mapper)

@CheckReturnValue 는 반환값을 확인한다는 의미.

@SchedulerSupport(value=“none”) 은 스케쥴러를 지원하지 않는다는 의미. 즉 현재 스레드에서 실행한다.

Function<T, R> 은 T 를 받아서 R 을 반환한다.

3.2. flatMap() 함수

-

map() 함수는 입력값을 어떤 함수에 넣어 변환할 수 있는 1:1 함수이다.

flatMap() 함수는 결과가 Observable 로 나온다는 것이 다르다.

-

Function<String, Observable<String>> getDoubleDiamonds = ball -> {
    Observable.just(ball + “<>”, ball + “<>”);
};
String[] balls = {“1”, “3”, “5”};
Observable.fromArray(balls).flatMap(getDoubleDiamond).subscribe(Log::i);

결과는..

value = 1<>

value = 1<>

value = 3<>

value = 3<>

value = 5<>

value = 5<>

-

돼왕 주 : map 과 flatMap 의 차이를 정확히 이해하기 위해 한번 검색해보았다.

요지는 아래와 같다.

map -> 1 to 1 mapping 이 된다.

flatMap -> flatMap 은 return 이 Observable 이기 떄문에, 1 to 0 or more mapping 이 된다. 또한 observable 이기 때문에 async 로직을 태우기 좋다.

관련해서 아래 글을 한번 읽어보길 추천한다.

https://stackoverflow.com/questions/22847105/when-do-you-use-map-vs-flatmap-in-rxjava

추가로 marble diagram 을 통해 emit 순서가 다를 수 있음도 주의해 보자.

3.2.1. 구구단 만들기

-

for 문을 Observable 로 변환

Observable.range(1, 9).subscribe( row -> System.out.println( dan + " * " + row + " = " + dan * row) );

사용자 함수 정의하기

Function<Integer, Observable<String>> gugudan = dan -> Observable.range(1,9).map(row -> dan + " * " + row + " = " + dan * row);

Observable.just(dan).flatMap( gugudan ).subscribe(System.out::println);

-

어떤 프로그램을 만들 때 for, while, if 와 같은 제어문이 아닌 함수로 생각하는 것은 프로그래머에게 커다란 도전 과제이다.

-

flatMap() 함수를 좀 더 활용하기

@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Observable<R> flatMap( Function<? super T, ? extends ObservableSource<? extends R>> mapper)

ObservableSource 는 Observable, AsyncSubject, BehaviorSubject, ConnectableObservable 등이 공통으로 구현한 인터페이스이다.

Single 클래스에는 SingleSource 라는 별도 인터페이스가 존재한다.

-

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final <U, R> Observable<R> flatMap(Function<? super T, ? extends ObservableSource<? extends U>> mapper, BiFunction<? super T, ? super U, ? extends R> resultSelector)

첫번째 mapper 의 인자로 받은 T 와 그것의 결과로 나오는 U 를 기반으로 새로운 Observable 을 만드는 녀석이다.

3.3. filter() 함수

-

first(default) : Observable 의 첫번째 값 필터. 값 없이 완료되면 기본값 리턴

last(default)

take(N) : 최소 N 개 값만 가져옴

takeLast(N) : 마지막 N 개 값만 가져옴

skip(N) : 최초 N개 값을 건너뜀

skipLast(N)

3.4. reduce() 함수

-

reduce 는 발행한 데이터를 모두 사용하여 어떤 최종 결과 데이터를 합성할 때 활용한다.

-

Strings[] balls = { “1”, “3”, “5” };
Maybe<String> source = Observable.fromArray(balls).reduce( (ball1, ball2) -> ball2 + “(“ + ball1 + “)”);
source.subscribe(System.out::println);

return type 이 Maybe 인 이유는 결과값이 없을 수도 있기 때문

결과는..

5(3(1))

3.4.1. 데이터 쿼리하기

-

함수형 프로그래밍에서는 일종의 자료구조인 클래스를 새로 만들기보다 Pair 혹은 Tuple 같은 좀 더 일반화된 자료구조를 선호한다. 기존에 만들어진 함수를 활용하는 폭이 넓어지기 때문이다.

3.5. 마치며