RxJava 리액티브 연산자 입문

19 Jan, 2021

RxJava 리액티브 연산자 입문

ReactiveX 문서의 연산자 부분을 살펴보면 다양한 연산자(operator) 함수가 존재합니다.
리액티브 연산자의 특징은 언어 특성과 크게 연관이 없다는 것입니다. ReactiveX는 자바뿐만 아니라 자바스크립트, 닷넷, 스칼라, 클로저, 스위프트의 리액티브 연산자 목록을 함께 제공하기 때문입니다.
리액티브 연산자 분류

연산자
- 생성(Creating) 연산자 : Observable, Single 클래스 등으로 데이터의 흐름을 만들어내는 함수.
- 변환(Transforming) 연산자 : 어떤 입력을 받아서 원하는 출력 결과를 내는 전통적인 의미의 함수.
- 필터(Filter) 연산자 : 입력 데이터 중에 원하는 데이터만 걸러내는 함수.
- 합성(Combining) 연산자 : 생성, 변환, 필터 연산자가 주로 단일 Observable을 다룬다면, 합성 연산자는 여러 Observable을 조합하는 역할을 하는 함수.
- 오류 처리(Error Handling) 연산자 : 예외 처리를 담당하는 함수.
- 유틸리티(Utility) 연산자 : 주요 연산자로는 subscribeOn()과 observeOn() 등이 있으며, 비동기 프로그래밍을 지원
- 조건(Conditional) 연산자 : Observable의 흐름을 제어하는 역할.
- 수학과 집합형(Mathematical and Aggregate) 연산자 : 수학 함수와 연관 있는 연산자.
- 배압(Back pressure) 연산자 : 배압 이슈에 대응하는 연산자.

map() 함수

map() 함수는 입력값을 어떤 함수에 넣어서 원하는 값으로 변환하는 함수.

ex) String을 String으로 변환 / String을 Integer나 다른 객체로 변환
map() 함수는 입력 데이터와 그것을 변환해줄 함수를 이어주는 중개업자가 있다고 생각하자!

>> map() 함수의 마블 다이어그램

앞서 말했던 중개업자는 마블 다이어그램의 map{} 사이에 있습니다.

>> map() 함수의 예시

public void marbleDiagram() {
		String[] balls = {"1", "2", "3", "5"};
		
		Observable<String> source = Observable.fromArray(balls)
				.map(ball -> ball + "<>");
		
		source.subscribe(System.out::println);
}

>>1<>
>>2<>
>>3<>
>>5<>

>> map() 함수의 Example Code

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
		ObjectHelper.requireNonNull(mapper, "mapper is null");
    return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
}

>> map() 함수의 원형

@CheckReturnValue는 반환값을 확인한다는 의미입니다.
@SchedulerSupport는 스케줄러를 지원하지 않는다는 의미 → 현재 스레드에서 실행한다는 의미입니다.
map() 함수는 매개변수로 Fuction 인터페이스 타입을 받습니다.

Function<? super T, ? extends R> mapper

>> map() 함수의 인자

위 코드는 람다식인데, RxJava의 람다 표현식은 아래의 제네릭 함수형 인터페이스 중 하나로 나타냅니다.

Predicate<T> : t값을 받아서 참이나 거짓을 반환 ( boolean test(T t) )

Consumer<T> : t값을 받아서 처리하고 반환값은 없음 ( void accept(T t) )

Function<T, R> : t값을 받아서 결과 R 반환 ( R apply(T t) )

public void mapFunction() {
		Function<String, String> getDiamond = ball -> ball + "<>";
		
		String[] balls = {"1", "2", "3", "5"};
		
		Observable<String> source = Observable.fromArray(balls).map(getDiamond);
		
		source.subscribe(System.out::println);
}

>>1<>
>>2<>
>>3<>
>>5<>

>> 위 marbleDiagram() 예제를 Function 인터페이스를 이용해 분리

map() 함수는 입력 타입과 반환 타입이 반드시 같을 필요는 없습니다.

public void mappingType() {
		Function<String, Integer> ballToIndex = ball -> {
			switch(ball) {
				case "RED" : return 1;
				case "YELLOW" : return 2;
				case "GREEN" : return 3;
				case "BLUE" : return 5;
				default : return -1;
			}
		};
		
		String[] balls = {"RED", "YELLOW", "GREEN", "BLUE"};
		Observable<Integer> source = Observable.fromArray(balls).map(ballToIndex);
		source.subscribe(System.out::println);
}

>>1
>>2
>>3
>>5

flatMap() 함수

flatMap() 함수는 map() 함수를 좀 더 발전시킨 함수입니다.
map() 함수는 원하는 입력값을 어떤 함수에 넣어서 변환할 수 있는 일대일 함수
flatMap() 함수는 원하는 입력값을 어떤 함수에 넣더라도 결과가 Observable로 나오는 일대다 혹은 일대일 Observable 함수

>> flatMap() 함수의 마블 다이어그램

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

>> flatMap() 함수의 원형

@CheckReturnValue는 반환값을 확인한다는 의미입니다.
@SchedulerSupport는 스케줄러를 지원하지 않는다는 의미 → 현재 스레드에서 실행한다는 의미입니다.
위 원형을 압축해서 요약하면 Observable flatMap(Function <T, ObservableSource>)입니다. “T를 넣으면 여러 개의 R이 나오도록 매핑한다.”는 의미입니다.

public void marbleDiagram() {
		Function<String, Observable<String>> getDoubleDiamonds = 
				ball -> Observable.just(ball + "<>", ball + "<>");
				
		String[] balls = {"1", "3", "5"};
		
		Observable<String> source = Observable.fromArray(balls)
				.flatMap(getDoubleDiamonds);

		//Observable<String> source = Observable.fromArray(balls)
		//	.flatMap(ball -> Observable.just(ball + "<>", ball + "<>");
		
		source.subscribe(System.out::println);
}

>>1<>
>>1<>
>>3<>
>>3<>
>>5<>
>>5<>

>> Example Code

Observable Type인 source는 balls 배열 값을 가져온 후 getDoubleDiamonds() 함수를 활용해 flatMap() 함수를 호출합니다.
getDoubleDiamonds() 함수는 String을 넣으면 “여러 개의 String을 발행하는 Observable이 나오는구나”라고 생각하면 됩니다.
getDoubleDiamonds() 함수는 ball을 입력받아서 ball + ”<>“을 두 번 발행합니다.

구구단 만들기

public void plainJava() {
		Scanner in = new Scanner(System.in);
		System.out.println("Gugudan Input : ");
		
		int dan =  Integer.parseInt(in.nextLine());
		
		for(int row = 1; row <= 9; row++) {
			System.out.println(dan + " * " + row + " = " + dan * row);
		}
}

>>Gugudan Input : 7
>>7 * 1 = 7
>>7 * 2 = 14
>>7 * 3 = 21
>>7 * 4 = 28
>>7 * 5 = 35
>>7 * 6 = 42
>>7 * 7 = 49
>>7 * 8 = 56
>>7 * 9 = 63

>> Java코드로 작성한 구구단

Step 1 : for문을 Observable로 변환하기

public void reactiveV1() {
		Scanner in = new Scanner(System.in);
		System.out.println("Gugudan Input : ");
		
		int dan = Integer.parseInt(in.nextLine());
		
		Observable<Integer> source = Observable.range(1, 9);
		source.subscribe(row -> System.out.println(dan + " * " + row + " = " + dan * row));
}

>>Gugudan Input : 7
>>7 * 1 = 7
>>7 * 2 = 14
>>7 * 3 = 21
>>7 * 4 = 28
>>7 * 5 = 35
>>7 * 6 = 42
>>7 * 7 = 49
>>7 * 8 = 56
>>7 * 9 = 63

>> for문 Observable 변환

Observable.range(start, count)는 start 숫자부터 count 개수만큼의 숫자 값을 발행합니다.

Step 2 : 사용자 함수 정의하기

1개의 입력을 받아서 9개의 줄을 출력하는 함수가 필요합니다.
map() 함수는 1개를 입력받아서 1개의 결과만 만들어내므로 이 예제에는 적합하지 않습니다.
따라서, 한 개의 Integer 값을 입력하면 여러 개의 String을 발행하는 flatMap() 함수를 활용하기에 적합합니다.

public void reactiveV2() {
		Scanner in = new Scanner(System.in);
		System.out.println("Gugudan Input : ");
		
		int dan = Integer.parseInt(in.nextLine());
		
		Function<Integer, Observable<String>> gugudan = num -> 
			Observable.range(1, 9).map(row -> num + " * " + row + " = " + num * row);
		
		Observable<String> source = Observable.just(dan).flatMap(gugudan);
		
		source.subscribe(System.out::println);
}

>>Gugudan Input : 7
>>7 * 1 = 7
>>7 * 2 = 14
>>7 * 3 = 21
>>7 * 4 = 28
>>7 * 5 = 35
>>7 * 6 = 42
>>7 * 7 = 49
>>7 * 8 = 56
>>7 * 9 = 63

>> flatMap() 함수를 활용한 코드 변환

gugudan() 함수를 만들어 두면 subscribe() 메서드를 호출해 출력하기만 하면 됩니다. → 코드 재활용성 증대
디버깅 도구들을 추가해 예외 처리하기도 쉽습니다.

filter() 함수

filter() 함수는 Observable에서 원하는 데이터만 걸러내는 역할을 합니다.
필요없는 데이터는 제거하고 필요한 데이터만 filter() 함수를 통과

>> filter() 함수의 마블 다이어그램

위 다이어그램을 보면 filter() 함수는 오직 동그란 모양의 원만 통과시킵니다.

>> filter() 함수의 예시

위 예시를 보면 인자로 받은 x의 값이 10보다 큰 경우 통과됩니다.

public void marbleDiagram() {
		String[] objs = {"1 CIRCLE", "2 DIAMOND", "3 TRIANGLE", "4 DIAMOND", "5 CIRCLE", "6 HEXAGON"};
		
		Observable<String> source = Observable.fromArray(objs).filter(obj -> obj.endsWith("CIRCLE"));
		
		source.subscribe(System.out::println);
}

>>1 CIRCLE
>>5 CIRCLE

>> filter() 함수의 활용 예

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Observable<T> filter(Predicate<? super T> predicate) {}

>> filter() 함수의 원형

filter() 함수의 원형을 보면, 함수형 인터페이스인 Predicate를 인자로 넣습니다. Predicate는 진위 판별이라는 뜻이 있으며 boolean 값을 리턴하는 특수한 함수형 인터페이스입니다.

public void marbleDiagram() {
		String[] objs = {"1 CIRCLE", "2 DIAMOND", "3 TRIANGLE", "4 DIAMOND", "5 CIRCLE", "6 HEXAGON"};

		Predicate<String> filterCircle = obj -> obj.endsWith("CIRCLE");
		
		Observable<String> source = Observable.fromArray(objs).filter(filterCircle);
		
		source.subscribe(System.out::println);
}

>>1 CIRCLE
>>5 CIRCLE

filter() 함수와 비슷한 함수들
1. first(default) 함수 : Observable의 첫 번째 값을 필터함. 만약 값없이 완료되면 대신 기본값을 리턴함.
2. last(default) 함수 : Observable의 마지막 값을 필터함. 만약 값없이 완료되면 대신 기본갑을 리턴함.
3. take(N) 함수 : 최초 N개 값만 가져옴.
4. takeLast(N) 함수 : 마지막 N개 값만 가져옴.
5. skip(N) 함수 : 최초 N개 값을 건너뜀.
6. skipLast(N) 함수 : 마지막 N개 값을 건너뜀.

public void otherFilters() {
		Integer[] numbers = {100, 200, 300, 400, 500};
		
		Single<Integer> single;
		Observable<Integer> source;
		
		//1. first
		single = Observable.fromArray(numbers).first(-1);
		single.subscribe(data -> System.out.println("first() value = " + data));
		
		//2. last
		single = Observable.fromArray(numbers).last(999);
		single.subscribe(data -> System.out.println("last() value = " + data));
		
		//3. take(N)
		source = Observable.fromArray(numbers).take(3);
		source.subscribe(data -> System.out.println("take(3) values = " + data));
		
		//4. takeLast(N)
		source = Observable.fromArray(numbers).takeLast(3);
		source.subscribe(data -> System.out.println("takeLast(3) values = " + data));
		
		//5. skip(N)
		source = Observable.fromArray(numbers).skip(2);
		source.subscribe(data -> System.out.println("skip(2) values = " + data));
		
		//6. skipLast(N)
		source = Observable.fromArray(numbers).skipLast(2);
		source.subscribe(data -> System.out.println("skipLast(2) values = " + data));
}

>>first() value = 100
>>last() value = 500
>>take(3) values = 100
>>take(3) values = 200
>>take(3) values = 300
>>takeLast(3) values = 300
>>takeLast(3) values = 400
>>takeLast(3) values = 500
>>skip(2) values = 300
>>skip(2) values = 400
>>skip(2) values = 500
>>skipLast(2) values = 100
>>skipLast(2) values = 200
>>skipLast(2) values = 300

>> filter() 함수와 비슷한 함수들의 활용 예

reduce() 함수

reduce() 함수는 발행한 데이터를 모두 사용하여 어떤 최종 결과 데이터를 합성할 때 활용합니다.
함수형 프로그래밍의 가장 기본 연산자인 map/filter/reduce 패턴을 이루는 마지막 필수 함수입니다.

>> reduce() 함수의 마블 다이어그램

>> reduce() 함수의 예시

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

>> 5(3(1))

>> reduce() 함수의 활용 예

reduce() 함수를 호출하면 인자로 넘긴 람다 표현식에 의해 결과 없이 완료될 수도 있습니다. 따라서 Observable이 아니라 Maybe 객체로 리턴됩니다.

Maybe 클래스

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

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Maybe<T> reduce(BiFunction<T, T, T> reducer) {}

>> reduce() 함수의 원형

reduce() 함수의 원형을 보면, BiFunction 인터페이스(첫 번째 인자로 받은 T와 그것의 결과로 나오는 두 번째 T를 기반으로 새로운 Observable을 만드는 인터페이스)를 인자로 활용합니다.

public void marbleDiagram() {
		String[] balls = {"1", "3", "5"};

		BiFunction<String, String, String> mergeBalls = (ball1, ball2) -> ball2 + "(" + ball1 + ")";
		
		Maybe<String> source = Observable.fromArray(balls)
				.reduce(mergeBalls);
		
		source.subscribe(System.out::println);
}

>> 5(3(1))

>> reduce() 함수의 람다 표현식 분리

데이터 쿼리하기

map(), filter(), reduce() 함수를 활용하여 간단한 데이터 쿼리 예제 만들기.

상점에서 발생한 오늘 매출

 - TV : $2,500
 - Camera : $300
 - TV : $1,600
 - Phone : $800

오늘 발생한 TV 매출의 총합은?

1. 전체 매출 데이터를 입력함.
2. 매출 데이터 중 TV 매출을 필터링함.
3. TV 매출의 합을 구함.

public void run() {
		List<Pair<String, Integer>> sales = new ArrayList<>();
		
		sales.add(Pair.create("TV", 2500));
		sales.add(Pair.create("Camera", 300));
		sales.add(Pair.create("TV", 1600));
		sales.add(Pair.create("Phone", 800));
		
		Maybe<Integer> tvSales = Observable.fromIterable(sales)
				.filter(sale -> "TV".equals(sale.getKey()))
						.map(sale -> sale.getValue())
						.reduce((sale1, sale2) -> sale1 + sale2);
		
		tvSales.subscribe(tot -> System.out.println("TV Sales : $" + tot));
}

>> TV Sales : $4100

>> TV 매출의 총합을 구하는 예제

key로 String, value로 Integer인 Pair 타입의 List 객체를 정의합니다.
각각 TV, Camera, Phone 제품의 매출액을 List 객체에 추가합니다.
TV의 매출액은 int 타입이므로 Maybe로 정의합니다.

Observable이 아니라 Maybe 클래스를 사용하는 이유는 결과 계산을 위한 reduce() 함수를 호출하기 위해서.
TV의 매출을 취합하기 위해 filter 함수를 활용합니다.
Pair 클래스에서 왼쪽 값(Key)을 얻기 위해서는 getLeft()/getKey() 메서드를, 오른쪽 값을 얻기 위해서는 getRight()/getValue() 메서드를 호출합니다.
map() 함수를 호출하여 매출액만 매핑합니다.
reduce() 함수를 활용하여 두 값을 더해줍니다.