安卓-kotlin协程的异常处理机制分析

作者

大家好，我叫?

本人于2020年10月加入37手游安卓团队

目前主要负责国内相关业务开发和一些日常业务

背景

使用kotlin的协程一段时间了，常用的用法也已经很熟悉，但都是停留在使用的阶段，没有对代码深入了解过，还是感觉有点虚；趁着过年这段时间，针对协程的异常处理，对其相关的源码学习了一波，梳理总结一下自己的理解。

本文基于 Kotlin v1.4.0，Kotlin-Coroutines v1.3.9源码分析

1、CoroutineScope源码分析

作用：创建和追踪协程，管理不同协程之间的父子关系和结构
创建协程的方式：

1、通过CoroutineScope创建

2、在协程中创建

第一种方式，首先如何通过CoroutineScope创建？

val scope = CoroutineScope(Job() + Dispatchers.Main) 

@Suppress("FunctionName")
public fun CoroutineScope(context: CoroutineContext): CoroutineScope =
    ContextScope(if (context[Job] != null) context else context + Job()) //没有job实例的话就搞一个

internal class ContextScope(context: CoroutineContext) : CoroutineScope {
    override val coroutineContext: CoroutineContext = context
}

CoroutineScope是一个全局的方法，然后在里面通过ContextScope就可以实例出来一个CoroutineScope对象了。
类似我们平时用到的MainScope或者Android平台上viewModelScope和lifecycleScope(只不过在生命周期相关回调做了有些自动cancel的处理)
也是跑到这里来。另外scope初始化的时候会有生成一个job，起到跟踪的作用
这里需要注意的是GlobalScope和普通协程的CoroutineScope的区别，GlobalScope的 Job 是为空的,因为它的coroutineContext是EmptyCoroutineContext，是没有job的

有了scope之后，我们就可以通过launch创建一个协程了

val job = scope.launch {}

戳代码看看

public fun CoroutineScope.launch(

    context: CoroutineContext = EmptyCoroutineContext,

    start: CoroutineStart = CoroutineStart.DEFAULT,

    block: suspend CoroutineScope.() -> Unit

): Job {

    。。。省略代码。。。
    return coroutine
}

launch参数有三个，前两个参数先不不分析，第三个是一个带receiver的lambda参数（参考Kotlin 中的Receiver 是什么），默认的类型是CoroutineScope

val job = scope.launch {①/* this: CoroutineScope */
    // 新的协程会将 CoroutineScope 作为父级 ，在launch里面创建
    //因为launch是一个扩展方法, 所以上面例子中默认的receiver是this，所以以下两种写法一样。这里可以理解为这里是一个回调，句柄是CoroutineScop
    launch { /* ... */ }
    this.launch { 
     // 通过 ① 创建的新协程作为当前协程的父级    
     }
}

再看看CoroutineScope.launch的实现

public fun CoroutineScope.launch(

    context: CoroutineContext = EmptyCoroutineContext,

    start: CoroutineStart = CoroutineStart.DEFAULT,

    block: suspend CoroutineScope.() -> Unit

): Job {

    //这里是根据父级创建新的上下文（协程的父级上下文），然后给下面创建协程用，具体逻辑下面代码块分析
    val newContext = newCoroutineContext(context)
   //这里就是创建协程
    val coroutine = if (start.isLazy)
        //协程真正的上下文生成是以newContext作为父级上下文生成的
        LazyStandaloneCoroutine(newContext, block) else
        StandaloneCoroutine(newContext, active = true)
        //start里面就是创建job相关的，不同的coroutine实例有不同的生成job策略
    coroutine.start(start, coroutine, block)
    return coroutine
}

public actual fun CoroutineScope.newCoroutineContext(context: CoroutineContext): CoroutineContext {
    //原来这是一个CoroutineScope的扩展函数，coroutineContext其实就是拿到到了scope对象的成员，然后通过“+”就可以搞成了，下面会说“+”
    //可以理解为把一个context数据add到一个 context map数据组中，还有一些逻辑判断，先不管，反正拿到的是一个新的context map
    val combined = coroutineContext + context
    //测试环境会给一下id拿来调试用的
    val debug = if (DEBUG) combined + CoroutineId(COROUTINE_ID.incrementAndGet()) else combined
    return if (combined !== Dispatchers.Default && combined[ContinuationInterceptor] == null)
        debug + Dispatchers.Default else debug
}

“+” 如何相加的？这就涉及到的相关类
CoroutineContext： 所有上下文的接口
CombinedContext：上下文组合时生成的类
CoroutineContext.Element：大部分单个上下文实现的类，因为有的会直接实现CoroutineContext

public operator fun plus(context: CoroutineContext): CoroutineContext =
        //operator操作符重载的特性 eg:Job() + Dispatchers.IO + CoroutineName("test") 就会跑到这里来 
        if (context === EmptyCoroutineContext) this else // fast path -- avoid lambda creation
            //acc为加数，element为被加数
            context.fold(this) { acc, element ->
                val removed = acc.minusKey(element.key)
                if (removed === EmptyCoroutineContext) element else {
                    // make sure interceptor is always last in the context (and thus is fast to get when present)
                    val interceptor = removed[ContinuationInterceptor]
                    if (interceptor == null) CombinedContext(removed, element) else {
                        val left = removed.minusKey(ContinuationInterceptor)
                        if (left === EmptyCoroutineContext) CombinedContext(element, interceptor) else
                            CombinedContext(CombinedContext(left, element), interceptor)
                    }
                }
            }

可以理解为一个map（实际上是一个单链表，详细的可以参考Kotlin协程上下文CoroutineContext是如何可相加的），通过key来获取不同类型的数据，需要改变的话使用当前的CoroutineContext来创建一个新的CoroutineContext即可。
上面提到的val scope = CoroutineScope(Job() + Dispatchers.Main)

综和以上两个代码片段，我们可以知道，一个新建的协程CoroutineContext的元素组成

1、有一个元素job，控制协程的生命周期

2、剩余的元素会从CoroutineContext 的父级继承，该父级可能是另外一个协程或者创建该协程的 CoroutineScope

2、CoroutineScope的类型

2.1、协程作用域对异常传播的影响

类型：

作用分析：

说明：

C2-1发生异常的时候，C2-1->C2->C2-2->C2->C1->C3（包括里面的子协程）->C4

C3-1-1发生异常的时候,C3-1-1->C3-1-1-1，其他不受影响

C3-1-1-1发生异常的时候，C3-1-1-1->C3-1-1，其他不受影响

2.2、示意代码

1、C1和C2没有关系

GlobalScope.launch { //协程C1


    GlobalScope.launch {//协程C2
        //...
    }
}

2、C2和C3是C1的子协程，C2和C3异常会取消C1

GlobalScope.launch { //协程C1


    coroutineScoope {
         launch{}//协程C2

         launch{}//协程C3

    }



} 

3、C2和C3是C1的子协程，C2和C3异常不会取消C1

GlobalScope.launch { //协程C1


    supervisorScope {
         launch{}//协程C2

         launch{}//协程C3

    }



} 

2.3、举个?

eg1：

@Test

fun test()= runBlocking{

    val handler = CoroutineExceptionHandler { coroutineContext, exception ->

        println("CoroutineExceptionHandler got $exception  coroutineContext ${coroutineContext}")

    }



    val job = GlobalScope.launch(handler) {

        println("1")

        delay(1000)

        coroutineScope {
            println("2")

            val job2 = launch(handler) {

                throwErrorTest()

            }

            println("3")

            job2.join()

            println("4")

        }

    }

    job.join()



}

fun throwErrorTest(){

    throw Exception("error test")

}

输出结果：

如果是协同作用域，job2所在的协程发生异常，会把job取消(不会打印“4”)，而且异常是从job所在协程抛出来的

eg2:

@Test

fun test()= runBlocking{

    val handler = CoroutineExceptionHandler { coroutineContext, exception ->

        println("CoroutineExceptionHandler got $exception  coroutineContext ${coroutineContext}")

    }



    val job = GlobalScope.launch(handler) {

        println("1")

        delay(1000)

        supervisorScope {
            println("2")

            val job2 = launch(handler) {

                throwErrorTest()

            }

            println("3")

            job2.join()

            println("4")

        }

    }

    job.join()



}

fun throwErrorTest(){

    throw Exception("error test")

}

输出结果：

如果是主从作用域，job2所在的协程发生异常，不会把job取消(会打印“4”)，而且异常是job2所在协程抛出来的

3、协程中异常处理的流程源码分析

3.1、协程的三层包装

第一层：launch和async返回的job，封装了协程的状态，提供取消协程的接口，实例都是继承自AbstractCoroutine

第二层：编译器生成（cps）的SuspendLambda的子类，封装了协程的真正运算逻辑，继承自BaseContinuationImpl，其中completion属性就是协程的第一层包装

第三层：DispatchedContinuation，封装了线程调度逻辑，包含了协程的第二层包装
三层包装都实现了Continuation接口，通过代理模式将协程的各层包装组合在一起，每层负责不同的功能
运算逻辑在第二层BaseContinuationImpl的resumeWith()函数中的invokeSuspend运行

3.2、发生异常的入口

BaseContinuationImpl中的resumeWith(result: Result<Any?>)处理异常的逻辑，省略的部分代码

public final override fun resumeWith(result: Result) {
    val completion = completion!! // fail fast when trying to resume continuation without completion
    val outcome: Result =
    。。。其他代码。。。
        try {
            val outcome = invokeSuspend(param)
            if (outcome === COROUTINE_SUSPENDED) return
            Result.success(outcome)
        } catch (exception: Throwable) {
            Result.failure(exception)
            }
            。。。其他代码。。。
        completion.resumeWith(outcome)
        。。。其他代码。。。
    }

由以上代码分析可知

1、invokeSuspend(param)方法的具体实现是在编译的生成的，对应协程体的处理逻辑

2、当发生异常的时候，即outcome为Result.failure(exception)，具体调用在completion.resumeWith(outcome)里面，通过AbstractCoroutine.resumeWith(Result.failure(exception))进入到第三层包装中

继续跟踪 AbstractCoroutine.resumeWith(result: Result) -> JobSupport.makeCompletingOnce(proposedUpdate: Any?): Any? -> JobSupport.tryMakeCompleting(state: Any?, proposedUpdate: Any?): Any?->JobSupport.tryMakeCompletingSlowPath(state: Incomplete, proposedUpdate: Any?): Any?

在tryMakeCompletingSlowPath方法中

var notifyRootCause: Throwable? = null
synchronized(finishing) {
    //。。。其他代码。。。
    notifyRootCause = finishing.rootCause.takeIf { !wasCancelling }
}

// process cancelling notification here -- it cancels all the children _before_ we start to to wait them (sic!!!)
// 该情景下，notifyRootCause 的值为 exception
notifyRootCause?.let { notifyCancelling(list, it) }

// otherwise -- we have not children left (all were already cancelled?)
return finalizeFinishingState(finishing, proposedUpdate)
//。。。其他代码。。。

如果发生异常即notifyRootCause不为空的时候，调用notifyCancelling方法,主要是取消子协程

private fun notifyCancelling(list: NodeList, cause: Throwable) {
    // first cancel our own children
    onCancelling(cause)
    notifyHandlers>(list, cause)
    // then cancel parent
    cancelParent(cause) // tentative cancellation -- does not matter if there is no parent
}

另外一个方法finalizeFinishingState，主要是异常传递和处理的逻辑，关键代码如下

private fun finalizeFinishingState(state: Finishing, proposedUpdate: Any?): Any? {
    。。。其他代码。。。
    // Now handle the final exception
    if (finalException != null) {
        //异常的传递和处理逻辑,如果cancelParent(finalException)不处理异常的话，就由当前
        //协程处理handleJobException(finalException)（具体实现在StandaloneCoroutine类处理异常，下文会提到）
        val handled = cancelParent(finalException) || handleJobException(finalException)
        if (handled) (finalState as CompletedExceptionally).makeHandled()
    }

    。。其他代码。。。
    return finalState
}

/**
 * The method that is invoked when the job is cancelled to possibly propagate cancellation to the parent.
 * Returns `true` if the parent is responsible for handling the exception, `false` otherwise.
 *
 * Invariant: never returns `false` for instances of [CancellationException], otherwise such exception
 * may leak to the [CoroutineExceptionHandler].
 * 返回指是true的话，异常由父协程处理，false的话异常由所在的协程来处理
 */
private fun cancelParent(cause: Throwable): Boolean {
    // Is scoped coroutine -- don't propagate, will be rethrown
    /**
    * Returns `true` for scoped coroutines.
    * Scoped coroutine is a coroutine that is executed sequentially within the enclosing scope without any concurrency.
    * Scoped coroutines always handle any exception happened within -- they just rethrow it to the enclosing scope.
    * Examples of scoped coroutines are `coroutineScope`, `withTimeout` and `runBlocking`.
   */
   //如果isScopedCoroutine true的话，即coroutineScope是主从作用域的话，异常是会传到父协程
    if (isScopedCoroutine) return true
    
    //cause是CancellationException的话是正常的协程结束行为，不会取消父协程
    /* CancellationException is considered "normal" and parent usually is not cancelled when child produces it.
     * This allow parent to cancel its children (normally) without being cancelled itself, unless
     * child crashes and produce some other exception during its completion.
     */
    val isCancellation = cause is CancellationException
    val parent = parentHandle
    // No parent -- ignore CE, report other exceptions.
    if (parent === null || parent === NonDisposableHandle) {
        return isCancellation
    }
    // Notify parent but don't forget to check cancellation
    //childCancelled(cause)为false的话，异常不会传递到父协程
    //使用SupervisorJob和supervisorScope时，子协程出现未捕获异常时也不会影响父协程，
    //它们的原理是重写 childCancelled() 为override fun childCancelled(cause: Throwable): Boolean = false
    return parent.childCancelled(cause) || isCancellation
}

由以上代码可知

1、出现未捕获异常时，首先会取消所有子协程

2、异常属于 CancellationException 时，不会取消父协程

3、使用SupervisorJob和supervisorScope时，即主从作用域，发生异常不会取消父协程，异常由所在的协程处理

3.3、CoroutineExceptionHandler的是如何生效的

在AbstractCoroutine中，处理异常的逻辑是在JobSupport接口中，默认是空的实现。
protected open fun handleJobException(exception: Throwable): Boolean = false
具体的实现逻辑是在StandaloneCoroutine中（Builders.common.kt文件）

private open class StandaloneCoroutine(
    parentContext: CoroutineContext,
    active: Boolean
) : AbstractCoroutine(parentContext, active) {
    override fun handleJobException(exception: Throwable): Boolean {
        //处理异常的逻辑
        handleCoroutineException(context, exception)
        return true
    }

}

具体实现如下

//CoroutineExceptionHandlerImpl.kt
private val handlers: List = ServiceLoader.load(
        CoroutineExceptionHandler::class.java,
        CoroutineExceptionHandler::class.java.classLoader
).iterator().asSequence().toList()

internal actual fun handleCoroutineExceptionImpl(context: CoroutineContext, exception: Throwable) {
    // use additional extension handlers
    for (handler in handlers) {
        try {
            handler.handleException(context, exception)
        } catch (t: Throwable) {
            // Use thread's handler if custom handler failed to handle exception
            val currentThread = Thread.currentThread()
            currentThread.uncaughtExceptionHandler.uncaughtException(currentThread, handlerException(exception, t))
        }
    }
    // 调用当前线程的 uncaughtExceptionHandler 处理异常
    // use thread's handler
    val currentThread = Thread.currentThread()
    currentThread.uncaughtExceptionHandler.uncaughtException(currentThread, exception)
}

以上的处理逻辑可以简单的归纳为以下伪代码

class StandardCoroutine(context: CoroutineContext) : AbstractCoroutine(context) {
    override fun handleJobException(e: Throwable): Boolean {
        context[CoroutineExceptionHandler]?.handleException(context, e) ?:
                Thread.currentThread().let { it.uncaughtExceptionHandler.uncaughtException(it, e) }
        return true
    }
}

所以默认情况下，launch式协程对未捕获的异常只是打印异常堆栈信息，如果使用了 CoroutineExceptionHandler 的话，只会使用自定义的 CoroutineExceptionHandler 处理异常。

小结

1、协程默认的作用域是协同作用域，异常会传播到父协程处理，即coroutineScope或者CoroutineScope(Job())这种形式。

2、协程作用域如果是主从作用域，异常不会传播到父协程处理，即supervisorScope 或 CoroutineScope(SupervisorJob()) 这种形式，其关键是重写 childCancelled()=false。

3、协程处理异常的时候，如果自定义CoroutineExceptionHandler的话，则由其处理，否则交给系统处理。

最后，本文异常处理分析是从协程作用域为切入点进行的，看代码过程中也会学到一些kotlin巧妙的语法使用；另外只是大概的去分析了一下异常的处理主线逻辑，有些细节的还需要去继续学习，下次会进行更加详细的分析，希望本文对你有帮助，也欢迎一起交流学习。

欢迎交流

过程中有问题或者需要交流的同学，可以扫描二维码加好友，然后进群进行问题和技术的交流等；