[RxJava2] 예외처리2 - 재시도 : retry(), retryUntil(), retryWhen()

공감 및 댓글은 포스팅 하는데  아주아주 큰 힘이 됩니다!! 포스팅 내용이 찾아주신 분들께  도움이 되길 바라며 더 깔끔하고 좋은 포스팅을  만들어 나가겠습니다^^

이번 포스팅에서는 예외처리 두 번째

재시도 함수들에 대해 공부합니다.

서버와 통신을 할 때 인터넷이 일시적으로 안되거나 서버에 일시적으로 장애가

발생하면 클라이언트에서는 일정 시간 후에 다시 통신을 요청하는 기능을

구현해놓을 필요가 있습니다.

이 때 1개의 API가 아닌 다수의 API 를 연속하여 호출해야 하는 시나리오가 있다면

꽤나 복잡해질 수도 있겠죠?

RxJava2에서는 이런 것들을 단순하게 처리할 수 있도록

retryXXX() 함수들을 제공합니다.

retry()

Observable에서 onError() 이벤트 발생 시 바로 다시 subscribe()를 호출해

재구독하도록 되어 있음.

예제)

package rx.java.chapter07;
 
import java.io.IOException;
 
import com.yudong80.reactivejava.common.Log;
 
import io.reactivex.Observable;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
 
public class RetryEx {
 
    static OkHttpClient client = new OkHttpClient();
    static String url = "https://api.github.com/zen";
 
    public static void main(String[] args) {
 
        final int RETRY = 5;
 
        Observable<String> src = Observable.just(url).map(RetryEx::get).retry((count, e) -> {
            System.out.println("count : " + count);
            try {
            Thread.sleep(1000);
            } catch (Exception ex) {
                // TODO: handle exception
            }
            return count < RETRY ? true : false;
        }).onErrorReturn(e -> "-500");
 
        src.subscribe(data -> System.out.println("result => " + data));
    }
 
    public static String get(String url) throws IOException {
        Request request = new Request.Builder().url(url).build();
        try {
            Response res = client.newCall(request).execute();
            return res.body().string();
        } catch (IOException e) {
            Log.et(e.getMessage());
            throw e;
        }
    }
 
}
 

onError() 발생시 바로 재시도가 되면 의미가 없으므로 1초의 Thread.sleep()을

주었습니다. 이 예제는 인터넷이 안되는 상황에서 실행해야 합니다 ㅎㅎ

인터넷이 되는 상황이면 정상동작합니다.

retryUntil()

특정 조건이 충족될 때까지 재시도하는 함수

retry() 는 재시도록 지속할 조건이 없을 때 재시도를 중단하고,

retryUntil() 은 재시도 중단 조건이 발생할 때까지 재시도함.

package rx.java.chapter07;
 
import java.io.IOException;
import java.net.InetAddress;
 
import com.yudong80.reactivejava.common.Log;
 
import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
 
public class RetryUntilEx {
 
    static OkHttpClient client = new OkHttpClient();
    static String url = "https://api.github.com/zen";
 
    public static void main(String[] args) {
        
        Observable<String> src = Observable.just(url)
                .map(RetryUntilEx::get)
                .subscribeOn(Schedulers.io())
                .retryUntil(() -> {
                    if(isNetworkAvailable()) {
                        return true;
                    }
                    Thread.sleep(1000);
                    return false;
                });
 
        src.subscribe(data -> System.out.println("result => " + data));
        try {
            Thread.sleep(5000);
        } catch (Exception e) {
            // TODO: handle exception
        }
    }
 
    public static String get(String url) throws IOException {
        Request request = new Request.Builder().url(url).build();
        try {
            Response res = client.newCall(request).execute();
            return res.body().string();
        } catch (IOException e) {
            Log.et(e.getMessage());
            throw e;
        }
    }
    
    public static boolean isNetworkAvailable() { 
        try {
            return InetAddress.getByName("www.naver.com").isReachable(1000);
        } catch (IOException e) {
            Log.v("Network is not available");    
        }
        return false;
    }
    
}
 

isNetworkAvailable() 함수는 현재 네트워크가 사용 가능한 상태인지를 파악하기 위한

함수 입니다. 우리 모두가 알 고 있는 네이버에 접근시도를 해서 접근이 되면 true

안되면 false를 반환합니다.

동작은 역시 네트워크가 안되는 상황에서 하셔야 에러발생을 확인할 수 있습니다.

retryWhen()

재시도 조건을 동적으로 설정해야하는 복잡한 로직을 구현할 때 활용하면 좋습니다.

단, retryWhen()의 인자가 Observable<Throwable> 이고 이해하기 어려운 함수이므로

충분한 공부가 필요할 듯 합니다.

아래는 총 5회 재시도를 하고 1번 실패할 때마다 딜레이를 1초씩 늘려나가는

예제입니다.

package rx.java.chapter07;
 
import java.util.concurrent.TimeUnit;
 
import io.reactivex.Observable;
 
public class RetryWhenEx {
 
    public static void main(String[] args) {
 
 
        Observable.<String>create(emit -> {
            emit.onError(new RuntimeException("fails"));
        }).retryWhen(rw -> {
            return rw.zipWith(Observable.range(1, 5), (a, b) -> b)
                    .flatMap(m -> {
                        System.out.println("delay " + m + "seconds");
                        return Observable.timer(m, TimeUnit.SECONDS);
                    });
        }).blockingForEach(System.out::println);
        
    }
 
}
 

이상입니다.

다음 포스팅에서는 Observable이 데이터를 발행하는 속도와 옵저버가 데이터를 받아

처리하는 속도의 차이가 발생할 때 사용하는 흐름 제어 함수에 대해 공부합니다.