[RxJava] 2. Reactive 기본 연산자(Operator) - map, filter, reduce

Reactive Programming에서 꽃이라고 할 수 있는 Reactive Operator에 대해 알아보도록 하겠습니다. 만약 여러분들이 Java 8을 사용하고 있다면, 이러한 Reactive 연산자가 매우 익숙하실 수도 있습니다.

 

RxJava에서는 Reactive Programming에 맞춰, 이러한 Reactive Operator를 제공합니다. 그러나 여러분들이 아셔야할 것은 이런 Reactive 연산자가 단연 RxJava, Java 8에서와 같이 "Java"에서만 볼 수 있는 것은 아닙니다. Python에서도 Javascript에서도 Reactive Programming만 지원한다면 이러한 연산자는 어디서든 볼 수 있습니다.

 

이번 글에서는 RxJava에서 Reative Operator가 어떤식으로 존재하는지 알아볼 것입니다. 그 전에 Reactive 연산자를 제대로 공부하고 짚어보도록 하죠.

 

 

 

Reactive Operator

앞서 말했듯이 Reactive Operator는 단연 Java에서만 존재하는 것이 아니라 .NET, Scala, Swift 등 Reactive Programming을 지원하는 언어라면 어디서든 사용할 수 있습니다. 따라서 이번에 RxJava를 통해서 이러한 Reactive 연산자를 숙달한다면 다른 언어에서도 쉽게 이용할 수 있습니다.

 

간단하게 특징을 정리해봤습니다.

 

• 다양한 연산 함수의 존재
  
  -> map, filter, reduce 등 400개 이상의 연산 함수가 존재하는데, 대부분 map, filter, reduce에서 파생된 것들입니다.
  
  
• 언어의 특성과는 무관
  
  -> 대부분의 Reactive Programming을 지원하는 언어에서 이러한 함수 이름을 그대로 사용.

 

그렇다면 Reactive Operator에는 어떠한 것들이 있을까요?

 

연산자 종류	정의	예제
생성(Creating) 연산자	Observable, Single 클래스 등을 이용하여 데이터의 흐름을 만들어내는 함수	create(), just(), fromArray(), interval(), range(), timer() 등
변환 (Transforming) 연산자	입력을 받아서 원하는 출력을 내는 전통적인 의미의 함수	map(), flatmap() 등
필터(Filter) 연산자	입력 데이터 중 원하는 데이터를 골라내는 함수	filter(), first(), take() 등
결합(Combining) 연산자	두 개 이상의 입력된 데이터를 하나의 데이터로 통합하는 연산자	zip(), combineLatest(), Merge(), concat()
오류처리(Error Handling) 연산자	연산자 내에서 예외 처리 구현을 위한 함수	onErrorReturn(), onErrorResumeNext(), retry() 등
조건(Conditional) 연산자	Observable의 흐름을 제어하는 역할	amb(), takeUntil(), skipUntil(), all() 등
수학과 집합형 연산자	수학 함수와 연관있는 연산자	sum() 등
기타 (Other) 연산자	구독, 발행 등의 이벤트 처리 및 데이터의 숫자를 세는 특징별 연산자	subscribeOn(), observeOn(), count() 등

이렇 듯 RxJava에서 제공하는 수많은 연산자들이 존재합니다만 이를 조금 간추려서 이렇게 정리해 볼 수 있습니다. 그러나 기본적으로는 map, filter, reduce에서 대부분 파생된 연산자들이며 이들의 연산자들은 다음 포스트에서 추후 자세하게 다루고, 이번 포스트에서는 기본 연산자인 map, filter, reduce에 대해서 중점적으로 다뤄보도록 하겠습니다.

 

 

 

 

map

map은 입력된 데이터에 여러분들이 정의한 함수를 넣어 원하는 데이터로 변환하는 함수입니다. 가령 Java의 Collections를 예로 들면 이런식으로 나타낼 수 있습니다.

 

우리는 Point라는 클래스를 만들고 이를 ArrayList에 저장했습니다. Java 8에서는 Stream 클래스에서 Reactive 연산자를 제공하여 여기서 map을 사용할 수 있죠. 

 

RxJava에서는 Observable 클래스를 한 개 만들고, 이를 사용해 볼 수 있습니다.

 

외부에서 데이터가 계속 들어온다는 가정하에 Observable 클래스를 사용할 경우, 위와 같이 한 개의 포인트만 주어질 때는 반복문을 사용할 수 있습니다. 그러나 Point가 한 개가 아닌 리스트 통째로 가져오게 된다면 반복문을 사용하지 않고, fromIterable 함수를 이용해서 코드를 줄일 수도 있습니다.

 

그렇다면 이 연산자는 어떤식으로 동작하는 것일까요?

 

우리는 이를 이해하기 위해 RxJava 문서에서 제공하는 그림을 한 번 참고해보도록 하겠습니다. 그림에서 보면 동그라미는 우리가 가지고 있던 Point 객체의 원본 데이터입니다. 이를 map 함수를 이용해서 원하는 연산 함수를 넣고, 마름모 형태로 바꿔주는 것이죠.

 

핵심은 내가 원하는 함수를 리스트에 있는 데이터에 넣고, 이를 "재가공"한다는 것입니다.

 

• .map(point -> point.toString());
  
  -> map 원형 : public final <R> Observable<R> map(Function<? super T, ? extends R> maper)
  
  
• 내가 원하는 형태를 만들기 위해 함수를 정의할 수 있음.

map 원형을 자세히 보면, 어떤 함수를 인자로 받게 됩니다. Function은 인터페이스이기 때문에 원하는 함수를 Lambda 식으로 정의할 수 있다는 장점을 가지고 있습니다.

 

 

 

 

flatMap

그렇다면 이와 비슷하게 생긴 flatMap은 무엇일까요? 같은 map인데 왜 flatMap이라는 것이 존재할까요?

 

flatMap은 연산 결과가 반드시 Observable로 나온다는 것이 map과 다른점입니다. map 함수는 1:1 데이터 연산 함수라면 flatMap은 1:N, 1:1 Observable 함수입니다.

 

또한 map은 동기 방식이기 때문에 비동기와 같이 사용할 경우 그 효과를 보기 어려우며 비동기 작업으로 데이터를 처리하고자 하는 경우 map이 아닌 flatMap을 사용해야 합니다.

 

그림을 먼저 보면, 역시 동그라미가 원본 데이터이고 마름모가 변형된 결과의 데이터라고 했을 때 한 Object 데이터가 아닌 여러 Object의 데이터를 사용했을 때 좀 더 효율적인 것을 볼 수 있습니다.

 

예를 들면 2차원 배열, Map을 사용했을 때 효율적이라고 볼 수 있습니다. 서로 각기 다른 종류의 Object로 되어 있고, 이 두 개를 모두 처리하려면 map을 사용했을 때는 서로가 다른 두 데이터에 대해 map을 한 번씩 써줘야 합니다. 그러나 flatMap을 사용한다면 서로 각기 다른 Object라 할지라도 이를 flatMap 한 번으로 처리할 수 있는 것이죠.

 

RxJava에서 Observable 클래스를 사용하면 여러 개의 데이터를 발행할 수 있기 때문에 만약 각기 다른 종류의 데이터를 받아야 한다면 flatMap을 사용해야 하고, 여기에 배압을 고려한다면 Flowable까지 사용하면 좋겠군요.

 

정리해보면 다음과 같습니다.

 

• map: 1개의 데이터를 다른 값이나 타입으로 변환.
• flatMap: 1개의 값을 받아 여러 개의 데이터(Observable)로 확장

 

 

 

 

filter

데이터들 중에서 원하는 데이터를 고를 수 있는 filter 함수는 Predicate를 인자로 넣어 원하는 데이터만 끄집어내는 함수입니다.

 

0, 0에서 2, 2까지의 좌표를 가지고 있는 사각형에서 오른쪽 좌표 데이터만 가져오고 싶을 때 위와 같이 Point에서 y 값이 2인 값만 가져오게끔 filter 안에 함수를 정의하여 원하는 데이터만 끄집어 낼 수 있습니다.

 

여기서 filter는 Function 인터페이스가 아니라 Predicate 인터페이스를 사용하기 때문에 내가 정의한 코드가 함수인지 Predicate인지를 좀 더 간단하게 사용하고 싶다면 위 코드처럼 Lambda를 사용하는 것도 하나의 방법이 될 수 있습니다.

 

• first() : Observable의 첫 번째 값 반환
• last() : Observable의 마지막 값 반환
• take(N) : 처음부터 N개까지의 값 반환
• takeLast(N) : 마지막에서부터 N개 까지의 값 반환
• skip(N) : 처음부터 N개까지의 값을 건너뛰고 그 다음 값부터 반환
• skipLast(N) : 마지막 N개까지의 값을 건너뛰고 그 이전 값부터 반환

filter를 사용하지 않고 이미 RxJava에서 정의된 필터링이 몇 개 있는데, 이를 사용해서도 필터링 할 수 있습니다. 

 

필터링 함수는 여러 가지의 데이터 중 원하는 동그라미 데이터만 뽑아낸다고 생각하면 됩니다. 

 

 

 

 

reduce

마지막으로 reduce 연산자는 두 개 이상의 여러 데이터를 하나의 데이터로 취합하는 연산 함수입니다. 가령 예를 들어 발행된 데이터를 모두 취합하여 리스트로 만들고, 모든 데이터를 조건 취합해서 하나의 데이터로 합쳐줄 때 reduce를 사용할 수 있습니다.

 

RxJava에서 reduce를 사용하면 Observable이 아닌 Maybe 클래스로 반환하게 됩니다. 그 이유는 reduce 안쪽 람다 함수 구현시 결과 값이 반환되지 않을 수도 있기 때문에 Nullable과 호환되는 Maybe 클래스로 반환됩니다.

 

ArrayList가 가지고 있는 모든 포인트의 합을 reduce를 이용해서 구한 코드입니다. reduce는 map, filter와 다르게 함수의 인자를 최대 2개까지 설정할 수 있으며 이를 이용해서 하나의 결과로 취합합니다. 단, 이 때 결과는 원본 데이터의 타입과 똑같은 타입으로 반환되어야 합니다.

 

• reduce(BitFunction<T, T, T> reducer)
  
  -> reduce는 Function이 아닌 BitFunction 인터페이스를 인자로 활용 (입력 인자가 최대 2개까지 적용)
  
  
• Observable이 아닌 Maybe로 원형 반환

 

그림으로 보면 위와 같이 원본 데이터를 합치고 합치고 또 합쳐서 마지막으로는 최종 하나의 데이터로 취합하는 형태가 만들어지는 것이 reduce입니다.

 

 

 

 

마치며...

Reactive Programming의 기본이라고 할 수 있는 map, filter, reduce에 대해 알아봤습니다. RxJava에서는 이를 바탕으로 하여 수 백개의 연산자가 존재하는데, 우리는 이 중에서 기본적인 연산자만을 다룬 것입니다.

 

그러나 이런 기본 연산자만 다룰 줄 알아도 어느 정도 Reactive Programming에 접근할 수 있습니다. 다음 포스트에서는 RxJava가 가진 연산자 중 자주 사용하는 연산자에 대해 다뤄볼 것이지만 모든 연산자를 다 외워서 써야 할 필요는 없습니다.

 

위에서 잠깐 다른 언어의 Reactive Programming에 대해서도 이야기 했었는데, Rx의 시초는 이전 포스트에서도 얘기했던 것처럼 .NET Core가 그 시초이며 해당 엔지니어가 Netflix로 이직하면서 Java에 대한 Rx를 만들고 나서 RxJava가 시작했다는 것으로 유래됩니다. 따라서 현재는 대부분의 프로그래밍 언어들이 Rx를 제공하고 있고, 그 중에서도 Java, Kotlin, Scala, Python이 제가 알고 있는 Reactive 연산자를 제공하는 프로그래밍 언어로 알고 있습니다.

 

이러한 연산자는 Java 8에서도 기본 제공되며 Java 8에서는 Stream 클래스 하위 메소드로 제공되기 때문에 각각의 클래스, 컬렉션을 이러한 연산자로 사용하고자 할 경우, Stream 클래스로 변환하여 사용할 수 있습니다. 하지만 Stream 클래스는 Observable과는 달리 데이터를 관찰하는 역할을 수행하지 않기 때문에 만약 데이터를 관찰하면서 Reactive Programming의 요소까지 사용하고자 한다면 그 때부터 RxJava를 이용해보는 것을 권합니다

 

 

 

참고(이미지): reactivex.io/documentation