基于Kotlin中Flow扩展重试方法

news2025/2/12 11:35:15

在这里插入图片描述
最近项目中统一采用Kotlin的Flow来重构了网络请求相关代码。
目前的场景是，接口在请求的时候需要一个accessToken值，因为此值会过期或者不存在，需要刷新，因此最终方案是在使用Flow请求的时候先获取accessToken值然后再进行接口请求，而获取accessToken值的方法已经封装成了一个Flow并且做了缓存，因此最后需要使用flatMapConcat操作符来连接真正需要的接口请求，如果获取的accessToken无效，又需要回头重新执行，逻辑如下：

判断本地是否存在accessToken并且是否过期，不存在或者已过期则请求accessToken
请求对应的接口
如果返回结果中accessToken无效，则重试

Flow提供了retry和retryWhen两种扩展方法来做重试操作：

retry源码

public fun <T> Flow<T>.retry(
    retries: Long = Long.MAX_VALUE,
    predicate: suspend (cause: Throwable) -> Boolean = { true }
): Flow<T> {
    require(retries > 0) { "Expected positive amount of retries, but had $retries" }
    return retryWhen { cause, attempt -> attempt < retries && predicate(cause) }
}

retryWhen源码

public fun <T> Flow<T>.retryWhen(predicate: suspend FlowCollector<T>.(cause: Throwable, attempt: Long) -> Boolean): Flow<T> =
    flow {
        var attempt = 0L
        var shallRetry: Boolean
        do {
            shallRetry = false
            val cause = catchImpl(this)
            if (cause != null) {
                if (predicate(cause, attempt)) {
                    shallRetry = true
                    attempt++
                } else {
                    throw cause
                }
            }
        } while (shallRetry)
    }

但是，retry和retryWhen只能通过异常来判断，如果是通过返回结果来判断，就需要借助外部变量来处理了，因此基于源码扩展了方法retry，可以接收请求结果，从而通过请求结果来判断是否需要重试。

fun <T> Flow<T>.retry(
    retries: Long = Long.MAX_VALUE, predicate: suspend (result: T) -> Boolean = { true }
): Flow<T> {
    require(retries > 0) { "Expected positive amount of retries, but had $retries" }
    return flow {
        var attempt = 0L
        var shallRetry: Boolean
        do {
            shallRetry = false
            try {
                collect {
                    if (attempt < retries && predicate(it)) {
                        shallRetry = true
                        attempt++
                    } else {
                        this.emit(it)
                    }
                }
            } catch (e: Throwable) {
                throw e
            }
        } while (shallRetry)
    }
}

最后的请求示例代码如下：

MainScope().launch {
        getToken().flatMapConcat {
            if (it is Result.Success) {
                sendMobileCode()
            } else {
                emptyFlow()
            }
        }.retry(1) {
            return@retry (it is Result.Failure) && (it.code == ErrorStatus.ACCESS_TOKEN_ERROR)
        }.flowOn(Dispatchers.IO).onStart {
            callback?.onStart()
        }.catch {
            callback?.onError(it)
        }.onCompletion {
            callback?.onComplete(it)
        }.collectLatest { result ->
        }
    }