Help us understand the problem. What is going on with this article?

RxJavaのエラーハンドリング

More than 1 year has passed since last update.

RxJavaでは非同期処理などで非常に便利なライブラリですが、そのエラーハンドリングもとてもスマートに行うことができます。ここではRxJavaのエラーハンドリングの詳細について説明したいと思います。

subscribeでのエラー処理

通常はsubscribe()でエラーを受け取ってハンドリングすることとなると思います。

subscribeでのエラー処理
        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        // エラー時の処理をここで受け取る
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

> subscribe
> on next: emit 1
> on next: emit 2
> error:java.lang.Throwable

注意が必要なのは下記のようにonNext()のハンドリングだけを行うようにしていると、onError()発火時にクラッシュします。

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        log("on next: " + s);
                    }
                });
> subscribe
> on next: emit 1
> on next: emit 2

Crash!! 
by rx.exceptions.OnErrorNotImplementedException

のでエラーが発生する可能性がある場合はきちんとハンドリングしましょう。

Errorをキャッチする

onErrorReturn

Observableチェインのなかで発生したErrorをキャッチして、大体のObjectに変換することでsubscriberにErrorが渡されるのを防ぐことができます。

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .onErrorReturn(new Func1<Throwable, String>() {
                    @Override
                    public String call(Throwable throwable) {
                        return "return";
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

> subscribe
> on next: emit 1
> on next: emit 2
> on next: return
> completed

エラー時に無事return文字列に変換されました。

onErrorResumeNext

Observableチェインのなかで発生したErrorをキャッチして、その中で再度Observableを呼び出すことで、エラー時に代替のStreamを返すことができます。

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .onErrorResumeNext(new Func1<Throwable, Observable<? extends String>>() {
                    @Override
                    public Observable<? extends String> call(Throwable throwable) {
                        return Observable.from(new String[]{ "resume 1", "resume 2"});
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

> subscribe
> on next: emit 1
> on next: emit 2
> on next: resume 1
> on next: resume 2
> completed

エラー時にresume 1,resume 2のストリームに変換されました。

エラー時にリトライする

retry

retry()メソッドはErrorが起こった時に、自動的にsubscribeしなおしてくれる、非常に便利なメソッドです。

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .retry()
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

> subscribe
> on next: emit 1
> on next: emit 2
> subscribe
> on next: emit 1
> on next: emit 2
> subscribe
> on next: emit 1
> on next: emit 2

...成功するまでリトライし続ける。。。

引数なしのretry()メソッドは、それまでの処理を成功するまでリトライし続けます。
これでは復旧できない問題が起こった時に無限ループになりかねないので、リトライの回数を制限するメソッドもあります。

// count: リトライ回数
public final Observable<T> retry(long count);

また、Errorの状況をみてリトライするかどうかも判断できます。

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .retry(new Func2<Integer, Throwable, Boolean>() {
                    @Override
                    public Boolean call(Integer count, Throwable throwable) {
                    // 2回までは無条件にリトライする
                        if(count < 3){
                            return true;
                        }
                    // それ以降はIllegalStateExceptionの場合だけリトライする
                        return throwable instanceof IllegalStateException;
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });
> subscribe
> on next: emit 1
> on next: emit 2
> subscribe
> on next: emit 1
> on next: emit 2
> subscribe
> on next: emit 1
> on next: emit 2
> error:java.lang.Throwable

retryWhen

retryWhen()メソッドはちょっと複雑でわかりにくいですが、より細かくリトライ処理を制御するためのメソッドです。retryWhen()のコールバックは初回からリトライ後に起こるErrorをObservableとして受け取り、それを変換して何らかのObservableとして返します。そしてこのコールバックで返したObservableで何らかの値がonNext()されると、そのタイミングでリトライされます。また、このObservableでonComplete()もしくはonError()を呼び出されると、もとのObservableがonComplete()もしくはonError()終了します。

ちょっとわかりにくいので簡単に例を示します。

3秒後にリトライ

3秒後にリトライするには以下のように記述します。

エラー発生後3秒待ってからリトライする
        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                    @Override
                    public Observable<?> call(final Observable<? extends Throwable> observable) {
                        return observable.flatMap(new Func1<Throwable, Observable<?>>() {
                            @Override
                            public Observable<?> call(Throwable throwable) {
                                return Observable.timer(3, TimeUnit.SECONDS);
                            }
                        });
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });
> subscribe
> on next: emit 1
> on next: emit 2

// 3秒後

> subscribe
> on next: emit 1
> on next: emit 2

// 3秒後
> subscribe
> on next: emit 1
> on next: emit 2

// 3秒後
> subscribe
> on next: emit 1
> on next: emit 2

// 3秒後
> subscribe
> on next: emit 1
> on next: emit 2

....繰り返し

Errorのまま終了

Errorのまま終了するには

Errorのまま終了する
        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                    @Override
                    public Observable<?> call(final Observable<? extends Throwable> observable) {
                        return observable.flatMap(new Func1<Throwable, Observable<?>>() {
                            @Override
                            public Observable<?> call(Throwable throwable) {
                                return Observable.error(throwable);
                            }
                        });
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });
> subscribe
> on next: emit 1
> on next: emit 2
> error:java.lang.Throwable

このコードはちょうどretryWhenメソッドを適用せずに、Errorを受け取ったのと同じ動作となります。

ErrorをハンドリングせずにComplete

Errorをハンドリングせずにその場でCompleteするには一見以下のコードで良さそうですが、これでは実は正しく動作しません。

Errorが来たらempty()でcompleteしたい
        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                    @Override
                    public Observable<?> call(final Observable<? extends Throwable> observable) {
                        return observable.flatMap(new Func1<Throwable, Observable<?>>() {
                            @Override
                            public Observable<?> call(Throwable throwable) {
                                return Observable.empty();
                            }
                        });
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });
> subscribe
> on next: emit 1
> on next: emit 2

詳しくはここのやり取りにもあるのですが、ここでflatMapしてempty()で返しても、もともとObservable<Throwable>の流れをcompleteするわけではないので、ここでは無視されてしまいます。

実際、Errorが起こったらいきなりcompleteしたい処理を書きたい場合には、実際には以下のように、errorのstreamを最初の一つだけ受け取って、それを無視する、というコードを書く必要があります。難しいですね。

Errorが来たらempty()でcompleteしたい
        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                    @Override
                    public Observable<?> call(final Observable<? extends Throwable> observable) {
                        return observable.take(1).ignoreElements();
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

> subscribe
> on next: emit 1
> on next: emit 2
> completed

ちゃんとCompleteできた!!

ちなみにtake(0)だとerrorが起こる前にいきなり全体のObservableが終了してしまいます。

余談ですが、エラー時に何もせずcompletedにしたい場合はonErrorResumeNext()を使ったほうがスマートかも知れません。

         .onErrorResumeNext(new Func1<Throwable, Observable<? extends String>>() {
                    @Override
                    public Observable<? extends String> call(Throwable throwable) {
                        return Observable.empty();
                    }
                })

3回リトライしたら終了

上記の例をもとに3回リトライする場合は、

3回リトライ
        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                    @Override
                    public Observable<?> call(final Observable<? extends Throwable> observable) {
                        return observable.take(3);
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

のように書きます。

> subscribe
> on next: emit 1
> on next: emit 2
> subscribe
> on next: emit 1
> on next: emit 2
> subscribe
> on next: emit 1
> on next: emit 2
> completed

ちゃんと3回リトライして終了しました。
この場合は先程のretry(count)のメソッドとの違いは、retry(count)ではリトライ回数が上限に達した後にerrorで終了するが、このケースではcompletedで終了する、という点にあります。

3秒後にリトライ、を3回試して終了

上記の例と、retryWhenの一番最初の例を組み合わせることで、3秒後にリトライ、を3回試して終了、という処理は以下のように書けます。

3秒後にリトライ、を3回試して終了
        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                    @Override
                    public Observable<?> call(final Observable<? extends Throwable> observable) {
                        return observable.take(3).flatMap(new Func1<Throwable, Observable<?>>() {
                            @Override
                            public Observable<?> call(Throwable throwable) {
                                return Observable.timer(3, TimeUnit.SECONDS);
                            }
                        });
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });


> subscribe
> on next: emit 1
> on next: emit 2

// 3秒後

> subscribe
> on next: emit 1
> on next: emit 2

// 3秒後

> subscribe
> on next: emit 1
> on next: emit 2
> completed

以上です。retryWhen()を駆使すればもっと細かなリトライ処理を書けそうですね!

あとがき

Advent Calendarの4日目としてエントリさせて頂いたのですが、いきなり期日に間に合わず、穴を開ける形となってしまいました。大変申し訳ありません。

このエントリを書く際に非常に困ったことがありました。初めはこのエントリのサンプルコードはすべてKotlinで書く予定だったんです。しかし、実際にためしてみるとKotlinではretryWhenのコードがコンパイル通らない!

以下のコードがコンパイルがとおらない

Observable
            .just("hoge")
            .retryWhen { it.flatMap{ Observable.timer(3, TimeUnit.SECONDS) } }

いろいろ調べたのですが、結局解決至らず、Kotlinでサンプルコードを書くのは諦めました。 orz

sansan
法人向け名刺管理サービスSansan及び個人向け名刺管理サービスEightを企画・開発・販売するベンチャー
http://jp.corp-sansan.com/
Why not register and get more from Qiita?
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
Comments
No comments
Sign up for free and join this conversation.
If you already have a Qiita account
Why do not you register as a user and use Qiita more conveniently?
You need to log in to use this function. Qiita can be used more conveniently after logging in.
You seem to be reading articles frequently this month. Qiita can be used more conveniently after logging in.
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
ユーザーは見つかりませんでした