RxJava map: 값 하나를 값 하나로 바꾸기

map은 원본 Flowable 또는 Observable의 각 onNext 값에 함수를 적용해 새 값 하나를 내보낸다. 입력 항목을 건너뛰거나 한 항목에서 여러 항목을 내보내는 연산자가 아니다. 건너뛰기는 filter, 여러 값을 내는 내부 스트림은 flatMap·concatMap으로 표현한다.

그림의 윗줄 원 하나가 변환 상자를 지나 아랫줄의 원 하나가 된다. 값의 형태는 바뀔 수 있어도 정상 통지 개수는 유지된다. 가운데 함수가 실패하면 그 시점에 오류로 끝나므로 그 이후 값은 나오지 않는다.

변환 결과 추적하기

import io.reactivex.Flowable;

public class MapExample {
    public static void main(String[] args) {
        Flowable.just("A", "B", "C")
            .map(String::toLowerCase)
            .subscribe(
                value -> System.out.println("값 " + value),
                Throwable::printStackTrace,
                () -> System.out.println("완료")
            );
    }
}
// 예상 출력: 값 a, 값 b, 값 c, 완료

just가 구독될 때 A, B, C를 내보내면 map은 순서대로 소문자를 만들고 구독자에게 전달한다. 별도 스케줄러가 없으므로 이 예제는 구독을 호출한 스레드에서 동기적으로 진행된다. map이 비동기 작업을 자동으로 병렬 처리하지 않는다. subscribeOn이나 observeOn을 넣으면 어느 구간에서 실행되는지 달라질 수 있지만 각 항목의 변환 계약은 같다.

실패와 제외를 구분하기

Flowable.just("10", "xx", "30")
    .map(Integer::parseInt)
    .subscribe(
        System.out::println,
        error -> System.out.println(error.getClass().getSimpleName())
    );
// 10, NumberFormatException. "30"은 통지되지 않음

xx를 숫자로 바꾸는 과정에서 예외가 나면 onError가 호출되고 원본의 뒤 값 30을 더 이상 처리하지 않는다. 변환 함수가 null을 반환해도 RxJava 2의 값 계약을 위반하므로 오류다. null을 “제외” 표시로 쓰지 않는다.

Flowable.just("10", "", "30")
    .filter(value -> !value.isEmpty())
    .map(Integer::parseInt)
    .subscribe(System.out::println); // 10, 30

빈 문자열을 무시하는 정책이라면 먼저 filter한다. 반면 xx 같은 잘못된 형식도 무시할지, 전체 입력을 실패로 할지는 업무 규칙으로 정한다. 모든 오류를 조용히 제외하면 데이터 품질 문제가 보이지 않는다.

map 안에 오래 걸리는 I/O를 넣으면 그 작업이 끝날 때까지 해당 실행 스레드가 점유된다. 여러 요청을 함께 실행해야 한다면 flatMap으로 작업 스트림을 만들 수 있지만 결과 순서·동시 실행 수·오류 처리 방식이 달라진다. 단순하고 빠른 일대일 변환에는 map을 유지하는 편이 예측하기 쉽다. 구독 취소 후에는 남은 원본 값이 더 전달되지 않을 수 있으며 이미 실행 중인 외부 호출의 중단은 그 호출 API에 달려 있다.

참고: ReactiveX Map, RxJava 2 Flowable API