Kotlin Coroutines(協程) 完全解析(四),協程的異常處理

收藏待读

Kotlin Coroutines(協程) 完全解析(四),協程的異常處理

Kotlin Coroutines(協程) 完全解析系列:

Kotlin Coroutines(協程) 完全解析(一),協程簡介

Kotlin Coroutines(協程) 完全解析(二),深入理解協程的掛起、恢復與調度

Kotlin Coroutines(協程) 完全解析(三),封裝異步回調、協程間關係及協程的取消

Kotlin Coroutines(協程) 完全解析(四),協程的異常處理

Kotlin Coroutines(協程) 完全解析(五),協程的並發

本文基於 Kotlin v1.3.0-rc-146,Kotlin-Coroutines v1.0.0-RC1

在上一篇文章中提到子協程拋出未捕獲的異常時默認會取消其父協程,而拋出 CancellationException 卻會當作正常的協程結束不會取消其父協程。本文來詳細解析協程中的異常處理,拋出未捕獲異常後協程結束後運行會不會崩潰,可以攔截協程的未捕獲異常嗎,如何讓子協程的異常不影響父協程。

Kotlin 官網文檔中有關於協程異常處理的文章,裏面的內容本文就不再重複,所以讀者們先閱讀官方文檔:

Coroutine Exception handling

協程的異常處理(官方文檔中文版)

看完官方文檔後,可能還是會有一些疑問:

  • launch 式協程的未捕獲異常為什麼會自動傳播到父協程,為什麼對異常只是在控制台打印而已?

  • async 式協程的未捕獲異常為什麼需要依賴用戶來最終消耗異常?

  • 自定義的 CoroutineExceptionHandler 的是如何生效的?

  • 異常的聚合是怎麼處理的?

  • SupervisorJobsupervisorScope 實現異常單向傳播的原理是什麼?

這些疑問在本文逐步解析協程中異常處理的流程時,會一一解答。

1. 協程中異常處理的流程

從拋出異常的地方開始跟蹤協程中異常處理的流程,拋出異常時一般都在協程的運算邏輯中。而在第二篇 深入理解協程的掛起、恢復與調度 中提到在協程的三層包裝中,運算邏輯在第二層 BaseContinuationImplresumeWith() 函數中的 invokeSuspend 運行,所以再來看一次:

internal abstract class BaseContinuationImpl(
    public val completion: Continuation?
) : Continuation, CoroutineStackFrame, Serializable {
    public final override fun resumeWith(result:Result) {
        ...
        var param = result
        while (true) {
            with(current) {
                val completion = completion!!
                val outcome: Result =
                    try {
                        // 調用 invokeSuspend 方法執行,執行協程的真正運算邏輯
                        val outcome = invokeSuspend(param)
                        // 協程掛起時 invokeSuspend 才會返回 COROUTINE_SUSPENDED,所以協程掛起時,其實只是協程的 resumeWith 運行邏輯執行完成,再次調用 resumeWith 時,協程掛起點之後的邏輯才能繼續執行
                        if (outcome === COROUTINE_SUSPENDED) return
                        Result.success(outcome)
                    } catch (exception: Throwable) {
                        // 注意這個 catch 語句,其實協程運算中所有異常都會在這裡被捕獲,然後作為一種運算結果
                        Result.failure(exception)
                    }
                releaseIntercepted() // this state machine instance is terminating
                if (completion is BaseContinuationImpl) {
                    // unrolling recursion via loop
                    current = completion
                    param = outcome
                } else {
                    // 這裡實際調用的是其父類 AbstractCoroutine 的 resumeWith 方法,當捕獲到異常時,調用 resumeWith(Result.failure(exception)) 更新協程狀態
                    completion.resumeWith(outcome)
                    return
                }
            }
        }
    }
}

從上面源碼的 try {} catch {} 語句來看,首先 協程運算過程中所有未捕獲異常其實都會在第二層包裝中被捕獲 ,然後會通過 AbstractCoroutine.resumeWith(Result.failure(exception)) 進入到第三層包裝中,所以協程的第三層包裝不僅維護協程的狀態,還處理協程運算中的未捕獲異常。這在第三篇分析子協程拋出未捕獲異常,默認情況會取消其父線程時也提到過。

繼續跟蹤 AbstractCoroutine.resumeWith(Result.failure(exception)) -> JobSupport.makeCompletingOnce(CompletedExceptionally(exception), defaultResumeMode) -> JobSupport.tryMakeCompleting(state, CompletedExceptionally(exception), defaultResumeMode),在最後 tryMakeCompleting() 過程中部分關鍵代碼:

private fun tryMakeCompleting(state:Any?, proposedUpdate:Any?, mode:Int): Int {
    ...
    // process cancelling notification here -- it cancels all the children _before_ we start to to wait them (sic!!!)
// 該情景下,notifyRootCause 的值為 exception
    notifyRootCause?.let { notifyCancelling(list, it) }
    // now wait for children
    val child = firstChild(state)
    if (child != null && tryWaitForChild(finishing, child, proposedUpdate))
        return COMPLETING_WAITING_CHILDREN
    // otherwise -- we have not children left (all were already cancelled?)
// 已取消所有子協程後,更新該協程的最終狀態
    if (tryFinalizeFinishingState(finishing, proposedUpdate, mode))
        return COMPLETING_COMPLETED
    // otherwise retry
    return COMPLETING_RETRY
}

先看 notifyCancelling(state.list, exception) 函數:

private fun notifyCancelling(list:NodeList, cause:Throwable) {
    // first cancel our own children
    onCancellation(cause)
// 這裡會調用 handle 節點的 invoke() 方法取消子協程,具體點就是調用 childJob.parentCancelled(job) 取消子協程
    notifyHandlers<JobCancellingNode>(list, cause)
    // then cancel parent
// 然後可能會取消父協程
    cancelParent(cause) // tentative cancellation -- does not matter if there is no parent
}

private fun cancelParent(cause:Throwable): Boolean {
    // CancellationException is considered "normal" and parent 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.
// CancellationException 是正常的協程結束行為,手動拋出 CancellationException 也不會取消父協程
    if (cause is CancellationException) return true
// cancelsParent 屬性也可以決定出現異常時是否取消父協程,不過一般該屬性都為 true
    if (!cancelsParent) return false
// parentHandle?.childCancelled(cause) 最後會通過調用 parentJob.childCancelled(cause) 取消父協程
    return parentHandle?.childCancelled(cause) == true
}

所以 出現未捕獲異常時,首先會取消所有子協程,然後可能會取消父協程。 而有些情況下並不會取消父協程,一是當異常屬於 CancellationException 時,而是使用 SupervisorJobsupervisorScope 時,子協程出現未捕獲異常時也不會影響父協程,它們的原理是重寫 childCancelled() 為 override fun childCancelled(cause: Throwable): Boolean = false

launch 式協程和 async 式協程都會自動向上傳播異常,取消父協程。

接下來再看 tryFinalizeFinishingState 的實現:

private fun tryFinalizeFinishingState(state:Finishing, proposedUpdate:Any?, mode:Int): Boolean {
    ...
// proposedException 即前面未捕獲的異常
    val proposedException = (proposedUpdate as? CompletedExceptionally)?.cause
    // Create the final exception and seal the state so that no more exceptions can be added
    var suppressed = false
    val finalException = synchronized(state) {
        val exceptions = state.sealLocked(proposedException)
        val finalCause = getFinalRootCause(state, exceptions)
        // Report suppressed exceptions if initial cause doesn't match final cause (due to JCE unwrapping)
// 如果在處理異常過程還有其他異常,這裡通過 finalCause.addSuppressedThrowable(exception) 的方式記錄下來
        if (finalCause != null) suppressed = suppressExceptions(finalCause, exceptions) || finalCause !== state.rootCause
        finalCause
    }
    ...
    // Now handle exception if parent can't handle it
// 如果 finalException 不是 CancellationException,而且有父協程且不為 SupervisorJob 和 supervisorScope,cancelParent(finalException) 都返回 true
// 也就是說一般情況下出現未捕獲的異常,一般會傳遞到最根部的協程,由最頂端的協程去處理
    if (finalException != null && !cancelParent(finalException)) {
        handleJobException(finalException)
    }
    ...
    // And process all post-completion actions
    completeStateFinalization(state, finalState, mode, suppressed)
    return true
}

上面代碼中 if (finalException != null && !cancelParent(finalException)) 語句可以看出,除非是 SupervisorJob 和 supervisorScope,一般協程出現未捕獲異常時,不僅會取消父協程,一步步取消到最根部的協程,而且最後還由最根部的協程(Root Coroutine)處理協程。下面繼續看處理異常的 handleJobException 的實現:

// JobSupport
protected open fun handleJobException(exception:Throwable) {}

// Builders.common.kt
private open class StandaloneCoroutine(
    parentContext: CoroutineContext,
    active: Boolean
) : AbstractCoroutine(parentContext, active) {
    override val cancelsParent: Boolean get() = true
    override fun handleJobException(exception:Throwable) = handleExceptionViaHandler(parentContext, exception)
}

// Actor
private open class ActorCoroutine(
    ...
) : ChannelCoroutine(parentContext, channel, active), ActorScope {
    override fun onCancellation(cause:Throwable?) {
        _channel.cancel(cause)
    }

    override val cancelsParent: Boolean get() = true
    override fun handleJobException(exception:Throwable) = handleExceptionViaHandler(parentContext, exception)
}

默認的 handleJobException 的實現為空,所以如果 Root Coroutine 為 async 式協程,不會有任何異常打印操作,也不會 crash,但是為 launch 式協程或者 actor 式協程的話,會調用 handleExceptionViaHandler() 處理異常。

下面接着看 handleExceptionViaHandler() 的實現:

internal fun handleExceptionViaHandler(context: CoroutineContext, exception: Throwable) {
    // Invoke exception handler from the context if present
    try {
        context[CoroutineExceptionHandler]?.let {
            it.handleException(context, exception)
// 如果協程有自定義 CoroutineExceptionHandler,則只調用 handler.handleException() 就返回
            return
        }
    } catch (t: Throwable) {
        handleCoroutineExceptionImpl(context, handlerException(exception, t))
        return
    }

    // If handler is not present in the context or exception was thrown, fallback to the global handler
// 如果沒有自定義 CoroutineExceptionHandler,
    handleCoroutineExceptionImpl(context, exception)
}

internal actual fun handleCoroutineExceptionImpl(context: CoroutineContext, exception: Throwable) {
    // use additional extension handlers
// 在 Android 中,還會有 uncaughtExceptionPreHandler 作為額外的 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))
        }
    }

    // use thread's handler
    val currentThread = Thread.currentThread()
// 調用當前線程的 uncaughtExceptionHandler 處理異常
    currentThread.uncaughtExceptionHandler.uncaughtException(currentThread, exception)
}

// Thread.java
public UncaughtExceptionHandler getUncaughtExceptionHandler() {
// 當前線程沒有定義 uncaughtExceptionHandler,會返回線程組作為 Thread.UncaughtExceptionHandler
    return uncaughtExceptionHandler != null ?
        uncaughtExceptionHandler : group;
}

// ThreadGroup.java
public void uncaughtException(Thread t, Throwable e) {
    if (parent != null) {
        parent.uncaughtException(t, e);
    } else {
        Thread.UncaughtExceptionHandler ueh =
            Thread.getDefaultUncaughtExceptionHandler();
// 優先使用線程通用的 DefaultUncaughtExceptionHandler,如果也沒有的話,則在控制台打印異常堆棧信息
        if (ueh != null) {
            ueh.uncaughtException(t, e);
        } else if (!(e instanceof ThreadDeath)) {
            System.err.print("Exception in thread ""
                                + t.getName() + "" ");
            e.printStackTrace(System.err);
        }
    }
}

所以默認情況下, launch 式協程對未捕獲的異常只是打印異常堆棧信息,如果在 Android 中還會調用 uncaughtExceptionPreHandler 處理異常。但是如果使用了 CoroutineExceptionHandler 的話,只會使用自定義的 CoroutineExceptionHandler 處理異常。

到這裡協程的異常處理流程就走完了,但是還有一個問題還沒解答, async 式協程的未捕獲異常只會導致取消自己和取消父協程,又是如何依賴用戶來最終消耗異常呢?

fun main(args:Array) = runBlocking {
    val deferred = GlobalScope.async {
        println("Throwing exception from async")
        throw IndexOutOfBoundsException()
    }
// await() 恢復調用者協程時會重寫拋出異常
    deferred.await()
}

看看反編譯的 class 文件就明白了:

public final Object invokeSuspend(@NotNull Object result){
    Object coroutine_suspended = IntrinsicsKt.getCOROUTINE_SUSPENDED();
    Deferred deferred;
    switch (this.label) {
        case 0:
            if (result instanceof Failure) {
                throw ((Failure) result).exception;
            }
            CoroutineScope coroutineScope = this.p$;
// 創建並啟動一個新的 async 協程
            deferred = BuildersKt.async$default((CoroutineScope) GlobalScope.INSTANCE, null, null, (Function2) new 1(null), 3, null);
            this.L$0 = deferred;
            this.label = 1;
// await() 掛起函數掛起當前協程,等待 async 協程的結果
            if (deferred.await(this) == coroutine_suspended) {
                return coroutine_suspended;
            }
            break;
        case 1:
            deferred = (Deferred) this.L$0;
// async 協程恢復當前協程時,傳遞進來的結果是 CompletedExceptionally(IndexOutOfBoundsException())
            if (result instanceof Failure) {
// 在當前協程重新拋出 IndexOutOfBoundsException 異常
                throw ((Failure) result).exception;
            }
            break;
        default:
            throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
    }
    return Unit.INSTANCE;
}

所以 async 式協程只有通過 await() 將異常重新拋出,不過可以可以通過 try { deffered.await() } catch () { ... } 來捕獲異常。

2. 小結

分析完協程的異常處理流程,其中需要注意的問題有下面這些:

  • 拋出 CancellationException 或者調用 cancel() 只會取消當前協程和子協程,不會取消父協程,也不會其他例如打印堆棧信息等的異常處理操作。

  • 拋出未捕獲的非 CancellationException 異常會取消子協程和自己,也會取消父協程,一直取消 root 協程,異常也會由 root 協程處理。

  • 如果使用了 SupervisorJob 或 supervisorScope,子協程拋出未捕獲的非 CancellationException 異常不會取消父協程,異常也會由子協程自己處理。

  • launch 式協程和 actor 式協程默認處理異常的方式只是打印堆棧信息,可以自定義 CoroutineExceptionHandler 來處理異常。

  • async 式協程本身不會處理異常,自定義 CoroutineExceptionHandler 也無效,但是會在 await() 恢復調用者協程時重新拋出異常。

原文 : Johnny Shieh

相關閱讀

免责声明:本文内容来源于Johnny Shieh,已注明原文出处和链接,文章观点不代表立场,如若侵犯到您的权益,或涉不实谣言,敬请向我们提出检举。