Wie man exponentielles Backoff wiederholt bei Kotlin-Coroutinen versucht

Question 1

Ich verwende Kotlin-Coroutinen für Netzwerkanfragen mit der Erweiterungsmethode, um die Klasse wie folgt nachzurüsten

public suspend fun <T : Any> Call<T>.await(): T {

  return suspendCancellableCoroutine { continuation -> 

    enqueue(object : Callback<T> {

        override fun onResponse(call: Call<T>?, response: Response<T?>) {
            if (response.isSuccessful) {
                val body = response.body()
                if (body == null) {
                    continuation.resumeWithException(
                            NullPointerException("Response body is null")
                    )
                } else {
                    continuation.resume(body)
                }
            } else {
                continuation.resumeWithException(HttpException(response))
            }
        }

        override fun onFailure(call: Call<T>, t: Throwable) {
            // Don't bother with resuming the continuation if it is already cancelled.
            if (continuation.isCancelled) return
            continuation.resumeWithException
        }
    })

      registerOnCompletion(continuation)
  }
}

dann verwende ich von der aufrufenden Seite die obige Methode wie folgt

private fun getArticles()  = launch(UI) {

    loading.value = true
    try {
        val networkResult = api.getArticle().await()
        articles.value =  networkResult

    }catch (e: Throwable){
        e.printStackTrace()
        message.value = e.message

    }finally {
        loading.value = false
    }

}

Ich möchte diesen API-Aufruf in einigen Fällen exponentiell wiederholen, dh (IOException), wie kann ich das erreichen?

Question 2

Ich würde vorschlagen, einen Helfer zu schreiben Funktion höherer Ordnung für Ihre Wiederholungslogik. Sie können die folgende Implementierung für den Anfang verwenden:

suspend fun <T> retryIO(
    times: Int = Int.MAX_VALUE,
    initialDelay: Long = 100, // 0.1 second
    maxDelay: Long = 1000,    // 1 second
    factor: Double = 2.0,
    block: suspend () -> T): T
{
    var currentDelay = initialDelay
    repeat(times - 1) {
        try {
            return block()
        } catch (e: IOException) {
            // you can log an error here and/or make a more finer-grained
            // analysis of the cause to see if retry is needed
        }
        delay(currentDelay)
        currentDelay = (currentDelay * factor).toLong().coerceAtMost(maxDelay)
    }
    return block() // last attempt
}

Die Verwendung dieser Funktion ist sehr einfach:

val networkResult = retryIO { api.getArticle().await() }

Sie können Wiederholungsparameter von Fall zu Fall ändern, zum Beispiel:

val networkResult = retryIO(times = 3) { api.doSomething().await() }

Sie können die Implementierung auch komplett ändern retryIO um den Anforderungen Ihrer Anwendung gerecht zu werden. Sie können beispielsweise alle Wiederholungsparameter fest codieren, die Begrenzung der Anzahl der Wiederholungen aufheben, die Standardeinstellungen ändern usw.

Question 3

Hier ein Beispiel mit der Flow und die retryWhen Funktion

RetryWhen Verlängerung :

fun <T> Flow<T>.retryWhen(
    @FloatRange(from = 0.0) initialDelay: Float = RETRY_INITIAL_DELAY,
    @FloatRange(from = 1.0) retryFactor: Float = RETRY_FACTOR_DELAY,
    predicate: suspend FlowCollector<T>.(cause: Throwable, attempt: Long, delay: Long) -> Boolean
): Flow<T> = this.retryWhen { cause, attempt ->
    val retryDelay = initialDelay * retryFactor.pow(attempt.toFloat())
    predicate(cause, attempt, retryDelay.toLong())
}

Verwendungszweck :

flow {
    ...
}.retryWhen { cause, attempt, delay ->
    delay(delay)
    ...
}

Question 4

Hier ist eine ausgefeiltere und bequemere Version meiner vorherigen Antwort, ich hoffe, es hilft jemandem:

class RetryOperation internal constructor(
    private val retries: Int,
    private val initialIntervalMilli: Long = 1000,
    private val retryStrategy: RetryStrategy = RetryStrategy.LINEAR,
    private val retry: suspend RetryOperation.() -> Unit
) {
    var tryNumber: Int = 0
        internal set

    suspend fun operationFailed() {
        tryNumber++
        if (tryNumber < retries) {
            delay(calculateDelay(tryNumber, initialIntervalMilli, retryStrategy))
            retry.invoke(this)
        }
    }
}

enum class RetryStrategy {
    CONSTANT, LINEAR, EXPONENTIAL
}

suspend fun retryOperation(
    retries: Int = 100,
    initialDelay: Long = 0,
    initialIntervalMilli: Long = 1000,
    retryStrategy: RetryStrategy = RetryStrategy.LINEAR,
    operation: suspend RetryOperation.() -> Unit
) {
    val retryOperation = RetryOperation(
        retries,
        initialIntervalMilli,
        retryStrategy,
        operation,
    )

    delay(initialDelay)

    operation.invoke(retryOperation)
}

internal fun calculateDelay(tryNumber: Int, initialIntervalMilli: Long, retryStrategy: RetryStrategy): Long {
    return when (retryStrategy) {
        RetryStrategy.CONSTANT -> initialIntervalMilli
        RetryStrategy.LINEAR -> initialIntervalMilli * tryNumber
        RetryStrategy.EXPONENTIAL -> 2.0.pow(tryNumber).toLong()
    }
}

Verwendungszweck:

coroutineScope.launch {
    retryOperation(3) {
        if (!tryStuff()) {
            Log.d(TAG, "Try number $tryNumber")
            operationFailed()
        }
    }
}

Question 5

Sie können diesen einfachen, aber sehr agilen Ansatz mit einfacher Verwendung ausprobieren:

BEARBEITEN: eine ausgefeiltere Lösung in einer separaten Antwort hinzugefügt.

class Completion(private val retry: (Completion) -> Unit) {
    fun operationFailed() {
        retry.invoke(this)
    }
}

fun retryOperation(retries: Int, 
                   dispatcher: CoroutineDispatcher = Dispatchers.Default, 
                   operation: Completion.() -> Unit
) {
    var tryNumber = 0

    val completion = Completion {
        tryNumber++
        if (tryNumber < retries) {
            GlobalScope.launch(dispatcher) {
                delay(TimeUnit.SECONDS.toMillis(tryNumber.toLong()))
                operation.invoke(it)
            }
        }
    }

    operation.invoke(completion)
}

Verwenden Sie es wie folgt:

retryOperation(3) {
    if (!tryStuff()) {
        // this will trigger a retry after tryNumber seconds
        operationFailed()
    }
}

Darauf kann man natürlich noch mehr aufbauen.

Question 6

Flow-Version https://github.com/hoc081098/FlowExt

package com.hoc081098.flowext

import kotlin.time.Duration
import kotlin.time.ExperimentalTime
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.FlowCollector
import kotlinx.coroutines.flow.emitAll
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.retryWhen

@ExperimentalTime
public fun <T> Flow<T>.retryWithExponentialBackoff(
  initialDelay: Duration,
  factor: Double,
  maxAttempt: Long = Long.MAX_VALUE,
  maxDelay: Duration = Duration.INFINITE,
  predicate: suspend (cause: Throwable) -> Boolean = { true }
): Flow<T> {
  require(maxAttempt > 0) { "Expected positive amount of maxAttempt, but had $maxAttempt" }
  return retryWhenWithExponentialBackoff(
    initialDelay = initialDelay,
    factor = factor,
    maxDelay = maxDelay
  ) { cause, attempt -> attempt < maxAttempt && predicate(cause) }
}

@ExperimentalTime
public fun <T> Flow<T>.retryWhenWithExponentialBackoff(
  initialDelay: Duration,
  factor: Double,
  maxDelay: Duration = Duration.INFINITE,
  predicate: suspend FlowCollector<T>.(cause: Throwable, attempt: Long) -> Boolean
): Flow<T> = flow {
  var currentDelay = initialDelay

  retryWhen { cause, attempt ->
    predicate(cause, attempt).also {
      if (it) {
        delay(currentDelay)
        currentDelay = (currentDelay * factor).coerceAtMost(maxDelay)
      }
    }
  }.let { emitAll(it) }
}