Kotlin协程之Flow异常示例处理

示例

代码如下：

launch(Dispatchers.Main) {
   // 第一部分
   flow {
       emit(1)
       throw NullPointerException("e")
   }.catch {
       Log.d("liduo", "onCreate1: $it")
   }.collect {
       Log.d("liudo", "onCreate2: $it")
   }
   // 第二部分
   flow {
       emit(1)
   }.onCompletion {
       Log.d("liduo", "onCreate3: $it")
   }.collect {
       Log.d("liudo", "onCreate4: $it")
   }
   // 第三部分
   flow {
       emit(1)
       throw NullPointerException("e")
   }.retryWhen { cause, attempt ->
       cause !is NullPointerException && attempt <= 2
   }.collect {
       Log.d("liudo", "onCreate5: $it")
   }
}

一.catch方法

catch方法用于捕获上游流产生的异常，代码如下：

public fun <T> Flow<T>.catch(action: suspend FlowCollector<T>.(cause: Throwable) -> Unit): Flow<T> =
   flow { // 创建Flow对象
       // 触发上游流的执行，并捕获异常
       val exception = catchImpl(this)
       // 捕获到异常，则回调action处理
       if (exception != null) action(exception)
   }

catch方法是Flow接口的扩展方法，并返回一个Flow类型的对象。在catch方法中，调用flow方法创建了一个Flow对象。

catch方法核心是通过catchImpl方法实现异常的捕获，如果成功捕获到异常，则回调参数action处理。这里参数action是FlowCollector接口的扩展方法，因此可以继续调用emit方法，向下游发送值。

catchImpl方法

当下游调用collect方法时，会触发catch方法创建的Flow对象的执行，并调用catchImpl方法来处理，代码如下：

internal suspend fun <T> Flow<T>.catchImpl(
   collector: FlowCollector<T>
): Throwable? {
   // 保存下游流执行抛出的异常
   var fromDownstream: Throwable? = null
   try {
       // 触发上游流的执行
       collect {
           try {
               // 将上游流发送的值作为参数，触发下游流执行
               collector.emit(it)
           } catch (e: Throwable) { // 如果下游流在执行中发生异常，保存并抛出
               fromDownstream = e
               throw e
           }
       }
   } catch (e: Throwable) { // 这里捕获的异常，可能为上游流的异常——collect方法，
                            // 也可能为下游流的异常——emit方法
       // 如果异常是下游流产生的异常，或者是协程取消时抛出的异常
       if (e.isSameExceptionAs(fromDownstream) || e.isCancellationCause(coroutineContext)) {
           throw e // 再次抛出，交给下游处理
       } else { // 如果是上游流的异常且不为协程取消异常
           return e // 成功捕获
       }
   }
   // 未捕获到异常，返回
   return null
}

catchImpl方法是Flow接口的扩展方法，因此在调用collect方法时，会触发上游流的执行。catchImpl方法的核心在于：将上游发出的值传递给下游处理，并对这一过程进行了异常捕获操作。

二. onCompletion方法

onCompletion方法用于在上游的流全部执行完毕后最后执行，代码如下：

public fun <T> Flow<T>.onCompletion(
   action: suspend FlowCollector<T>.(cause: Throwable?) -> Unit
): Flow<T> = unsafeFlow { // 创建一个Flow对象
   try {
       // 触发上游流的执行
       // this表示下游的FlowCollector
       collect(this)
   } catch (e: Throwable) {// 如果下游发生异常
       // 将异常封装成ThrowingCollector类型的FlowCollector，并回调参数action，
       ThrowingCollector(e).invokeSafely(action, e)
       // 抛出异常
       throw e
   }
   // 如果正常执行结束，会走到这里
   val sc = SafeCollector(this, currentCoroutineContext())
   try {
       // 回调执行参数action
       sc.action(null)
   } finally {
       sc.releaseIntercepted()
   }
}

onCompletion方法是Flow接口的扩展方法，因此在调用collect方法时，会触发上游流的执行。同时，传入this作为参数，this表示下游流调用collect方法时，传给unsafeFlow方法创建的Flow对象的类型为FlowCollector的对象。onCompletion方法的核心在于：将自身创建的Flow对象作为上游与下游的连接容器，只有当流全部执行完毕或执行过程中发生异常，collect方法才可以执行完成，继续向下执行。

1.unsafeFlow方法

unsafeFlow方法用于创建一个类型为Flow对象，与之前在Kotlin协程：Flow基础原理提到过的SafeFlow类相比，unsafeFlow方法创建的Flow对象不会对执行的上下文进行检查，代码如下：

@PublishedApi
internal inline fun <T> unsafeFlow(@BuilderInference crossinline block: suspend FlowCollector<T>.() -> Unit): Flow<T> {
   // 返回一个匿名内部类
   return object : Flow<T> {
       // 回调collect方法是直接执行block
       override suspend fun collect(collector: FlowCollector<T>) {
           collector.block()
       }
   }
}

虽然onCompletion方法内部使用unsafeFlow方法创建Flow对象，但却使用了SafeCollector类。根据之前在Kotlin协程：Flow基础原理提到的，调用SafeCollector类的emit方法时，会对上下文进行检查。因此实际效果与使用SafeFlow类效果相同。

2.ThrowingCollector类

ThrowingCollector类也是一种FlowCollector，用于包裹异常。当调用它的emit方法时，会抛出包裹的异常，代码如下：

private class ThrowingCollector(private val e: Throwable) : FlowCollector<Any?> {
   override suspend fun emit(value: Any?) {
       // 抛出异常
       throw e
   }
}

为什么要重新创建ThrowingCollector对象，而不使用下游的FlowCollector对象呢？

为了防止当下游的流执行失败时，onCompletion方法的action参数执行时调用emit方法发送数据，这样会导致onCompletion方法作为在&ldquo;finially代码块&rdquo;使用时不是最后执行的方法。

onCompletion方法搭配与catch方法，实现try-catch-finially代码块的效果。

三. retryWhen方法

retryWhen方法与catch方法类似，都可以用于捕获上游流产生的异常。但两者不同之处在于，retryWhen方法还可以根据&ldquo;异常类型&rdquo;和&ldquo;重试次数&rdquo;来决定是否要再次触发上游流的执行，而且当retryWhen方法不打算再次触发上游流的执行时，捕获的异常会被抛出，代码如下：

// 参数cause表示捕获到的异常
// 参数attempt表示重试的次数
// 参数predicate返回true表示重新触发上游流的执行
public fun <T> Flow<T>.retryWhen(predicate: suspend FlowCollector<T>.(cause: Throwable, attempt: Long) -> Boolean): Flow<T> =
   // 创建一个Flow对象
   flow {
       // 记录重试次数
       var attempt = 0L
       // 表示是否重新触发
       var shallRetry: Boolean
       do {
           // 复位成false
           shallRetry = false
           // 触发上游流的执行，并捕获异常
           val cause = catchImpl(this)
           // 如果捕获到异常
           if (cause != null) {
               // 用户判断，是否要重新触发
               if (predicate(cause, attempt)) {
                   // 表示要重新触发
                   shallRetry = true
                   // 重试次数加1
                   attempt++
               } else { // 如果用户不需要重新触发
                   // 则抛出异常
                   throw cause
               }
           }
       // 判断是否重新触发
       } while (shallRetry)
   }

retryWhen方法是Flow接口的扩展方法。retryWhen方法的核心通过catchImpl方法实现对上游流的触发及异常捕获，并加入了由用户判断的重试逻辑实现。

来源：https://blog.csdn.net/LeeDuoZuiShuai/article/details/126828155