9. RxJava 통지 데이터 변환 연산자 — 거꾸로 바라본 세상

이 페이지는 통지된 값을 다른 형태로 만드는 연산자들의 목차다. 입력 한 건을 출력 한 건으로 바꾸는 경우에는 map, 입력마다 새로운 스트림을 만드는 경우에는 flatMap이나 concatMap을 살펴본다.

map:       1 ── 2 ── 3  →  "1" ── "2" ── "3"
flatMap:   1 ── 2       →  [1의 비동기 작업] + [2의 비동기 작업]
buffer:    1 ── 2 ── 3  →  [1, 2, 3]

flatMap은 내부 스트림을 합치는 동안 결과 순서가 입력 순서와 달라질 수 있다. 순서를 유지해야 한다면 concatMap이 맞는지 확인한다. buffer는 값을 변환하기보다 여러 통지를 목록으로 묶는다. 각 링크의 예제에서 입력 개수, 출력 개수, 순서, 완료 시점을 비교하며 읽으면 차이가 분명해진다.

간단한 변환 예제

Flowable.range(1, 3)
        .map(number -> "상품-" + number)
        .subscribe(System.out::println);
// 상품-1, 상품-2, 상품-3

Flowable.range(1, 3)
        .concatMap(number -> Flowable.just(number, number * 10))
        .subscribe(System.out::println);
// 1, 10, 2, 20, 3, 30

첫 예는 값마다 문자열 하나를 만든다. 둘째 예는 입력마다 두 값을 내는 내부 스트림을 만들고 입력 순서대로 이어 붙인다. 내부 작업이 비동기로 완료되는 경우 flatMap은 먼저 끝난 작업의 결과를 먼저 전달할 수 있다. 요청 순서가 결과 순서와 같아야 하는 파일 처리라면 concatMap, 독립 작업을 병렬로 처리해도 된다면 flatMap을 검토한다.

오류 신호도 결과 흐름의 일부다. 내부 스트림 하나가 실패하면 기본 동작으로 전체 흐름이 실패할 수 있으므로, DelayError 변형을 사용할지는 실패를 모아 처리해도 되는 업무 규칙을 기준으로 결정한다. 큰 목록을 buffer로 묶을 때는 한 번에 메모리에 쌓이는 항목 수를 제한한다.

선택할 때 확인할 세 가지

연산자	한 입력이 만드는 결과	순서와 대기 비용
`map`	값 하나	입력 순서를 유지하며 비동기 경계를 만들지 않는다.
`flatMap`	내부 스트림의 0개 이상 값	내부 작업을 함께 진행할 수 있지만 결과가 섞일 수 있다.
`concatMap`	내부 스트림의 0개 이상 값	앞 작업의 완료를 기다려 순서를 지킨다.
`buffer`	여러 입력을 담은 목록 하나	묶음이 찰 때까지 출력이 늦어지고 목록만큼 메모리를 사용한다.

예를 들어 입력 세 개에 각각 두 값을 반환하는 concatMap은 여섯 값을 낸다. 같은 변환 함수를 map에 넣으면 Flowable 객체 세 개가 결과이며 그 안의 값을 자동으로 꺼내지 않는다. flatMap으로 바꾸면 여섯 값은 나오지만 비동기 내부 스트림 사이의 순서는 달라질 수 있다. 외부 호출처럼 부작용이 있는 작업에서는 재시도 시 어느 범위가 다시 실행되는지도 함께 확인한다.

concatMapDelayError는 여러 입력을 계속 처리한 뒤 오류를 내보내는 선택지다. 이미 실패한 입력까지 성공한 것처럼 만드는 기능은 아니다. 오류를 값으로 변환해야 하는 경우에는 어느 입력이 실패했는지 알 수 있는 결과 타입을 설계한다. 더 많은 값을 기다리며 버퍼에 담는 방법은 처리 속도 자체를 높여 주지 않으므로 느린 소비자 문제는 배압과 처리 용량의 관점에서 따로 판단한다.

참고: RxJava 2 Flowable API.