에러 처리 연산자

  RxJava에서는 아래와 같이 try ~ catch 문을 사용할 수 없습니다. 아래 코드를 실행하면 catch 에서 Exception을 잡지 못하고 예외가 발생하는 것을 확인할 수 있습니다.

package com.itvillage.chapter05.chapter0506;

import com.itvillage.utils.LogType;
import com.itvillage.utils.Logger;
import com.itvillage.utils.TimeUtil;
import io.reactivex.Observable;

import java.util.concurrent.TimeUnit;

public class CanNotUseTryCatchExample {
    public static void main(String[] args) {
        try{
            Observable.just(2)
                    .map(num -> num / 0)
                    .subscribe(System.out::println);
        } catch (Exception e){
            Logger.log(LogType.PRINT, "# 에러 처리가 필요: " + e.getCause());
        }
    }
}

 

  RxJava에서는 일반적으로 다음과 같이 에러 발생 시 Observable을 생성한 함수에서 onError()를 호출하고 subscribe의 onError()에서 해당 에러를 받아 처리하는 구조를 가집니다.

package com.itvillage.chapter05.chapter0506;

import com.itvillage.utils.LogType;
import com.itvillage.utils.Logger;
import com.itvillage.utils.TimeUtil;
import io.reactivex.Observable;

import java.util.concurrent.TimeUnit;

/**
 * 데이터를 처리하다가 예외가 발생할 경우 일반적인 예제
 * 0으로 나누는 부분에서 예외가 발생한다.
 *
 * RxJava에서 에러를 처리하는 일반적인 방식의 예제
 *  - RxJava에서는 에러 발생 시, Observable을 생성한 함수에서 onError()를 호출하고,
 *  - subscribe의 onError()에서 해당 error를 받아서 처리하는 구조를 가진다.
 */
public class GeneralErrorHandleExample {
    public static void main(String[] args) {
        Observable.just(5)
                .flatMap(num -> Observable
                        .interval(200L, TimeUnit.MILLISECONDS)
                        .doOnNext(data -> Logger.log(LogType.DO_ON_NEXT, data))
                        .take(5)
                        .map(i -> num / i))
                .subscribe(
                        data -> Logger.log(LogType.ON_NEXT, data),
                        error -> Logger.log(LogType.ON_ERROR, error),
                        () -> Logger.log(LogType.ON_COMPLETE)
                );

        TimeUtil.sleep(1000L);
    }
}

 

onErrorReturn

• 에러가 발생했을 때 에러를 의미하는 데이터로 대체할 수 있습니다.
• onErrorReturn()을 호출하면 onError 이벤트는 발생하지 않습니다.

package com.itvillage.chapter05.chapter0506;

import com.itvillage.utils.LogType;
import com.itvillage.utils.Logger;
import com.itvillage.utils.TimeUtil;
import io.reactivex.Observable;

import java.util.concurrent.TimeUnit;

/**
 * onErrorReturn()를 사용해 예외 발생 시, 우리가 원하는 값을 전달하는 예제
 * - 예외가 발생될 가능성이 있는 부분에 대해서 사전에 처리를 선언할 수 있다.
 * - 소비자가 예상되는 예외를 모두 사전에 알고 처리하긴 힘들기때문에 생산자쪽에서 예외 처리를 사전에 해두고 소비자는 선언된
 * 예외 상황을 보고 그에 맞는 적절한 처리를 할 수 있다.
 */
public class ObservableOnErrorReturnExample {
    public static void main(String[] args) {
        Observable.just(5)
                .flatMap(num -> Observable
                        .interval(200L, TimeUnit.MILLISECONDS)
                        .take(5)
                        .map(i -> num / i)
                        .onErrorReturn(exception -> {
                            if(exception instanceof ArithmeticException)
                                Logger.log(LogType.PRINT, "계산 처리 에러 발생: " + exception.getMessage());

                            return -1L;
                        })
                )
                .subscribe(
                        data -> {
                            if(data < 0)
                                Logger.log(LogType.PRINT, "# 예외를 알리는 데이터: " + data);
                            else
                                Logger.log(LogType.ON_NEXT, data);
                        },
                        error -> Logger.log(LogType.ON_ERROR, error),
                        () -> Logger.log(LogType.ON_COMPLETE)
                );

        TimeUtil.sleep(1000L);
    }
}

 

onErrorResumeNext

• 에러가 발생했을 때 에러를 의미하는 Observable로 대체할 수 있습니다.
• Observable로 대체할 수 있으므로 데이터 교체와 더불어 에러 처리를 위한 추가 작업을 할 수 있습니다.

package com.itvillage.chapter05.chapter0506;

import com.itvillage.utils.LogType;
import com.itvillage.utils.Logger;
import com.itvillage.utils.TimeUtil;
import io.reactivex.Observable;

import java.util.concurrent.TimeUnit;

/**
 * onErrorResumeNext를 이용해서 에러 발생시, 다른 Observable로 대체하는 예제.
 */
public class ObservableOnErrorResumeNextExample {
    public static void main(String[] args) {
        Observable.just(5L)
                .flatMap(num -> Observable
                        .interval(200L, TimeUnit.MILLISECONDS)
                        .take(5)
                        .map(i -> num / i)
                        .onErrorResumeNext(throwable -> {
                            Logger.log(LogType.PRINT, "# 운영자에게 이메일 발송: " + throwable.getMessage());
                            return Observable.interval(200L,TimeUnit.MILLISECONDS).take(5).skip(1).map(i -> num / i);
                        })
                ).subscribe(data -> Logger.log(LogType.ON_NEXT, data));

        TimeUtil.sleep(2000L);
    }
}

 

retry

• 데이터 통지 중 에러가 발생했을 때, 데이터 통지를 재시도 합니다.
• 즉, onError 이벤트가 발생하면 subscribe()를 다시 호출하여 재구독합니다.
• 에러가 발생한 시점에 통지에 실패한 데이터만 다시 통지되는 것이 아니라 처음부터 다시 통지됩니다.

package com.itvillage.chapter05.chapter0506;

import com.itvillage.utils.LogType;
import com.itvillage.utils.Logger;
import com.itvillage.utils.TimeUtil;
import io.reactivex.Observable;

import java.util.concurrent.TimeUnit;

/**
 * 에러 발생시 재시도를 즉시 하지 않고, 지연 시간을 주고 재시도를 하는 예제
 */
public class ObservableRetryExample02 {
    private final static int RETRY_MAX = 5;
    public static void main(String[] args) {
        Observable.just(5)
                .flatMap(
                        num -> Observable
                                .interval(200L, TimeUnit.MILLISECONDS)
                                .map(i -> {
                                    long result;
                                    try{
                                        result = num / i;
                                    }catch(ArithmeticException ex){
                                        Logger.log(LogType.PRINT, "error: " + ex.getMessage());
                                        throw ex;
                                    }
                                    return result;
                                })
                                .retry((retryCount, ex) -> {
                                    Logger.log(LogType.PRINT, "# 재시도 횟수: " + retryCount);
                                    TimeUtil.sleep(1000L);
                                    return retryCount < RETRY_MAX ? true : false;
                                })
                                .onErrorReturn(throwable -> -1L)

                ).subscribe(
                        data -> Logger.log(LogType.ON_NEXT, data),
                        error -> Logger.log(LogType.ON_ERROR, error),
                        () -> Logger.log(LogType.ON_COMPLETE)
                );


        TimeUtil.sleep(6000L);
    }
}

 

 

 

[참고 자료]

• 인프런- Kevin의 알기 쉬운 RxJava
• https://reactivex.io/documentation/operators.html
• 전체 소스코드