리액티브 프로그래밍 예외 처리

RxJava 스트림은 값 onNext, 정상 종료 onComplete, 오류 종료 onError를 구분한다. onError가 오면 그 구독은 끝나며 같은 소스에서 다음 값이 계속 오지 않는다. 예외를 일반 값 -1로 바꿀지, 전체 작업을 실패로 끝낼지, 다시 시도할지는 업무 규칙에 따라 정해야 한다.

flowchart LR
    A[작업 중 오류] --> B{대응 방식}
    B --> C[onError: 실패 전달]
    B --> D[retry: 소스 재구독]
    B --> E[onErrorReturnItem: 대체 값 후 완료]
    B --> F[onErrorResumeNext: 대체 스트림 후 완료]

그림에서 retry는 실패한 명령 한 줄만 다시 실행하는 것이 아니라 소스에 다시 구독할 수 있다. 앞에서 이미 관찰한 값이 중복으로 전달될 수 있으므로 파일 쓰기·결제 같은 부작용에는 그대로 적용하지 않는다. onErrorReturnItem은 오류를 정상 종료로 바꾸므로 실패 사실을 추적해야 할 경우 별도 기록이 필요하다.

구독자가 오류를 받는 경로

import io.reactivex.Flowable;

public class ErrorSignals {
    public static void main(String[] args) {
        Flowable.just(1, 0, 2)
            .map(n -> 10 / n)
            .subscribe(
                value -> System.out.println("값 " + value),
                error -> System.out.println("오류 " + error.getClass().getSimpleName()),
                () -> System.out.println("완료")
            );
        // 값 10, 오류 ArithmeticException
    }
}

n=0에서 나눗셈이 실패하면 2는 처리되지 않고 완료도 오지 않는다. 오류 소비자를 빼면 RxJava의 전역 오류 처리 경로에 의존하게 되어 애플리케이션에서 실패 원인을 놓치기 쉽다. 구독 경계에서 오류를 받고 요청·작업 ID와 함께 기록한다.

재시도는 몇 번 실행되는가

import io.reactivex.Flowable;
import java.util.concurrent.atomic.AtomicInteger;

public class RetryExample {
    public static void main(String[] args) {
        AtomicInteger attempts = new AtomicInteger();
        Flowable<Integer> source = Flowable.fromCallable(() -> {
            int attempt = attempts.incrementAndGet();
            if (attempt < 3) throw new IllegalStateException("일시 실패");
            return attempt;
        });
        source.retry(2).subscribe(
            value -> System.out.println("성공 " + value),
            error -> System.out.println("실패 " + error)
        );
        // 성공 3: 최초 1회 + 재시도 최대 2회
    }
}

retry(2)는 최초 시도 외에 최대 두 번 더 구독한다. 코드의 실패 조건은 세 번째 시도에서만 성공하므로 결과가 3이다. 원문의 .retry(2).retryUntil(...)처럼 재시도 연산자를 연달아 붙이면 각각의 재구독 경계를 계산하기 어려워진다. 조건부 재시도가 필요하면 예외 종류, 최대 횟수, 지연을 하나의 정책으로 정하고 해당 RxJava 버전의 API를 확인한다. 영구적인 입력 오류를 무한 재시도하면 작업과 외부 시스템을 계속 압박한다.

대체 값과 대체 스트림

Flowable.just(1, 0, 2)
    .map(n -> 100 / n)
    .onErrorReturnItem(-1)
    .subscribe(System.out::println);
// 100, -1 뒤에 완료. 원래 입력 2는 처리되지 않는다.

대체값 -1이 실제 정상 데이터와 겹칠 수 있다면 오류를 값으로 바꾸는 계약이 모호하다. onErrorResumeNext는 한 값이 아닌 다른 스트림으로 이어갈 때 사용한다. 캐시로 대체하는 읽기 작업에는 유용하지만 최신 데이터가 필요하거나 실패를 호출자에게 알려야 하면 조용한 대체가 위험하다. onExceptionResumeNext는 모든 Throwable을 똑같이 다루지 않으므로 심각한 오류까지 복구 대상으로 삼지 않도록 예외 종류를 확인한다.

재시도와 대체 모두 오류가 난 시점 이전의 부작용을 되돌리지 않는다. 외부 호출이 성공했지만 응답을 받기 전에 연결이 끊겼다면 재시도는 중복 요청이 될 수 있다. 멱등 키, 트랜잭션, 중복 제거가 필요한지 업무 흐름에서 결정한다.

참고: RxJava 2 Flowable API, ReactiveX Retry