2024/6/26

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

この記事からわかること

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

index

[open]

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

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

posted withアプリーチ

環境

Android Studio：Flamingo
Kotlin：1.8.20

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

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

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

retryメソッド

公式リファレンス：Retry

retryはストリームにエラーが発生した場合にストリームを最初から再試行するメソッドです。以下の場合は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: エラー発生

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

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

// 独自のエラー型を定義
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として受け取り、最終的に新しいObservableを生成することで再試行を制御しています。そのためretryWhenの中でストリームを操作するような処理(遅延や条件分岐など)を行うことができます。

例えば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)

まだまだ勉強中ですので間違っている点や至らぬ点がありましたら教えていただけると助かります。

ご覧いただきありがとうございました。

Android