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

做者

你们好，我叫🐜java

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

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

背景

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

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

一、CoroutineScope源码分析

做用：建立和追踪协程，管理不一样协程之间的父子关系和结构建立协程的方式：async

一、经过CoroutineScope建立ide

二、在协程中建立函数

第一种方式，首先如何经过CoroutineScope建立？oop

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的元素组成

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

二、剩余的元素会从CoroutineContext 的父级继承，该父级多是另一个协程或者建立该协程的 CoroutineScope

二、CoroutineScope的类型

2.一、协程做用域对异常传播的影响

类型：

做用分析：

说明：

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.二、示意代码

一、C1和C2没有关系

GlobalScope.launch { //协程C1
    GlobalScope.launch {//协程C2
        //...
    }
}
复制代码

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

GlobalScope.launch { //协程C1
    coroutineScoope {
         launch{}//协程C2
         launch{}//协程C3
    }
} 
复制代码

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

GlobalScope.launch { //协程C1
    supervisorScope {
         launch{}//协程C2
         launch{}//协程C3
    }
} 
复制代码

2.三、举个🌰

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.一、协程的三层包装

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

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

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

3.二、发生异常的入口

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)
        。。。其余代码。。。
    }
复制代码

由以上代码分析可知

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

二、当发生异常的时候，即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
}
复制代码

由以上代码可知

一、出现未捕获异常时，首先会取消全部子协程

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

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

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 处理异常。

小结

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

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

三、协程处理异常的时候，若是自定义CoroutineExceptionHandler的话，则由其处理，不然交给系统处理。

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

欢迎交流

过程当中有问题或者须要交流的同窗，能够扫描二维码加好友，而后进群进行问题和技术的交流等；