2024/6/26

【Kotlin/Android】RxJavaのretryやretryWhenの使い方！エラー時の再試行

この記事からわかること

Android Studio/KotlinでRxJavaの使い方
retryやretryWhenメソッドの違いと使い方

index

[open]

エラー発生時に処理をリトライしたい場合
retryメソッド
flatmapなどでは呼び出し位置に注意
エラーの種類によってリトライ条件を変更する
retryWhenメソッド

＼アプリをリリースしました／

友達や家族の誕生日をメモ！通知も届く-みんなの誕生日-

posted withアプリーチ

環境

Android Studio：Meerkat
Kotlin：2.0.21
Android OS：15以降

エラー発生時に処理をリトライしたい場合

RxJavaでストリームにエラーが発生した場合に処理をリトライするにはretryやretryWhenを使用します。エラーハンドリングするだけであればdoOnErrorやonErrorReturnでも可能です。

retry・・・エラーが発生した場合にそのまま再試行を行う演算子
retryWhen・・・エラーが発生した際にそのエラーを受け取り、新しいObservableを生成して再試行

retryメソッド

公式リファレンス：Retry

retryはストリームにエラーが発生した場合にストリームを最初から再試行するメソッドです。ObservableやSingle、Completableなど基本的にほとんどのストリーム生成するクラスで使用することが可能です。以下の場合は1、2が流れた後にエラーを発生させています。

val observable = Observable.create<Int> { emitter ->
    Log.d("RxJava", "ストリーム開始")
    emitter.onNext(1)
    emitter.onNext(2)
    emitter.onError(RuntimeException("エラー発生"))
}

observable
    .retry(3) // 最大3回まで再試行
    .doOnSubscribe { Log.d("RxJava", "観測開始") }
    .doOnNext { Log.d("RxJava", "Received: $it") }
    .doOnError { Log.d("RxJava", "Error: ${it.message}") }
    .doOnComplete { Log.d("RxJava", "Completed") }
    .subscribe(
        { /* onNext */ },
        { /* onError */ },
        { /* onComplete */ }
    )
    .addTo(compositeDisposable)

retryの引数には再試行したい回数か条件(後述)を渡します。再試行回数を指定しない場合は無限回のリトライになります。retryより下流にあるメソッドは実行されなくなるのでエラーが流れても12行目のdoOnErrorは動作しません。

出力

観測開始
ストリーム開始
Received: 1
Received: 2
ストリーム開始
Received: 1
Received: 2
ストリーム開始
Received: 1
Received: 2
ストリーム開始
Received: 1
Received: 2
Error: エラー発生

flatmapなどでは呼び出し位置に注意

retryや後述のretryWhenメソッドは呼び出した箇所から上流部分を再試行します。そのためflatmapなどで連結している場合は呼び出す位置によって再施行される範囲が変わってくるので注意してください。

例えばflatmapの後に設置した場合はflatmap内でエラーが発生しても最初のObservable.just(1)から処理が再施行されます。

Observable.just(1)
    .flatMap { id ->
        api.getData(id)
    }
    .retry(3) // ← 全体をリトライ
    .subscribe(...)

そのためapi.getDataだけを再施行したい場合は以下のようretryメソッドを設置します。

Observable.just(1)
    .flatMap { id ->
        api.getData(id)
            .retry(3) // ← 内側のObservableだけリトライ
    }
    .subscribe(...)

エラーの種類によってリトライ条件を変更する

発生したエラーの種類によってリトライしたい場合としたくない場合が分かれることは多いと思います。例えば以下のように独自のエラー型を定義しておりそのエラーが流れた時だけリトライさせることも可能です。

// 独自のエラー型を定義
class CustomException(message: String) : Exception(message)

val observable = Observable.create<Int> { emitter ->
    println("ストリーム開始")
    emitter.onNext(1)
    emitter.onNext(2)
    if (false) {
      emitter.onError(RuntimeException("エラー発生"))
    } else {
      emitter.onError(CustomException("カスタムエラー発生"))
    }
}

observable
    .retry { retryCount, throwable ->
        if (throwable is CustomException) {
            Log.d("RxJava", "発生したエラー： ${throwable.message}, 残りリトライ回数: $retryCount")
            // リトライ回数が3回未満であればリトライする
            retryCount < 3
        } else {
            // その他のエラーはリトライしない
            false
        }
    }
    .doOnSubscribe { Log.d("RxJava", "観測開始")}
    .doOnNext { Log.d("RxJava", "Received: $it") }
    .doOnError { Log.d("RxJava", "Error: ${it.message}") }
    .doOnComplete { Log.d("RxJava", "Completed") }
    .subscribe(
        { /* onNext */ },
        { /* onError */ },
        { /* onComplete */ }
    )
    .addTo(compositeDisposable)

再試行させるためのエラー分岐処理はretryメソッドにラムダ式を渡しその引数から以下の2つを参照できます。

1つ目：現在のリトライ回数を示す整数
2つ目：発生したエラー（例外）

あとは取得したエラーが期待するものであるかどうかを識別してfalseを返せば再試行をせず、trueを返すと再試行してくれます。

.retry { retryCount, throwable ->
    if (throwable is CustomException) {
        Log.d("RxJava", "発生したエラー： ${throwable.message}, 残りリトライ回数: $retryCount")
        // リトライ回数が3回未満であればリトライする
        retryCount < 3 // true
    } else {
        // その他のエラーはリトライしない
        false
    }
}

retryWhenメソッド

同じような再試行を行うメソッドにretryWhenがあります。retryWhenは受け取ったエラーをObservable / Flowableとして受け取り、最終的に新しいObservable / Flowableを生成することで再試行を制御しています。そのためretryWhenの中でストリームを操作するような処理(遅延や条件分岐など)を行うことができます。また記述していますがObservableまたはFlowableでしか使用できません。

例えばretryWhenを使用してストリームの再試行に2秒間のディレイを設けつつ、最大3回に制限したい場合は以下のように実装します。

val observable = Observable.create<Int> { emitter ->
    println("ストリーム開始")
    emitter.onNext(1)
    emitter.onNext(2)
    if (false) {
        emitter.onError(RuntimeException("エラー発生"))
    } else {
        emitter.onError(CustomException("カスタムエラー発生"))
    }
}

observable
    .retryWhen { errors ->
        // zipWithで2つのObservableを連結してペアとして受け取れる
        // Observable.range(1, 3)で1〜3を流すストリームを用意してリトライ回数とする
        errors.zipWith(Observable.range(1, 3)) { throwable, retryCount ->
            Log.d("RxJava", "発生したエラー： ${throwable.message}, 残りリトライ回数: $retryCount")
            throwable // 次のflatMapの引数で参照したい値を指定する
        }.flatMap { throwable ->
            // 2秒の遅延を挿入
            Observable.timer(2, TimeUnit.SECONDS)
        }
    }
    .doOnSubscribe { Log.d("RxJava", "観測開始")}
    .doOnNext { Log.d("RxJava", "Received: $it") }
    .doOnError { Log.d("RxJava", "Error: ${it.message}") }
    .doOnComplete { Log.d("RxJava", "Completed") }
    .subscribe(
        { /* onNext */ },
        { /* onError */ },
        { /* onComplete */ }
    )
    .addTo(compositeDisposable)