Kotlin協(xié)程啟動(dòng)createCoroutine及創(chuàng)建startCoroutine原理

更新時(shí)間：2022年08月04日 11:10:14 作者：xfhy

這篇文章主要為大家介紹了Kotlin協(xié)程啟動(dòng)createCoroutine及創(chuàng)建startCoroutine原理詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

createCoroutine 和 startCoroutine

協(xié)程到底是怎么創(chuàng)建和啟動(dòng)的？本篇文章帶你揭曉。

在Continuation.kt文件中，有2個(gè)基礎(chǔ)API，這里單獨(dú)提出來(lái)說(shuō)一下，方便后面我們理解launch。

public fun <T> (suspend () -> T).createCoroutine(
    completion: Continuation<T>
): Continuation<Unit> =
    SafeContinuation(createCoroutineUnintercepted(completion).intercepted(), COROUTINE_SUSPENDED)
public fun <T> (suspend () -> T).startCoroutine(
    completion: Continuation<T>
) {
    createCoroutineUnintercepted(completion).intercepted().resume(Unit)
}

createCoroutine和startCoroutine就是用來(lái)創(chuàng)建和啟動(dòng)協(xié)程的基礎(chǔ)API，launch、async等在底層一定程度上都使用了該基礎(chǔ)API，launch和async只不過(guò)是封裝而已。所以，我們先掌握它們。

這2個(gè)函數(shù)看起來(lái)差別不大，一個(gè)調(diào)用了resume開(kāi)始了協(xié)程，一個(gè)沒(méi)有調(diào)用，需要外部去調(diào)用resume（createCoroutine會(huì)把Continuation返回出去）。

既然launch和async可以用它們來(lái)創(chuàng)建和啟動(dòng)協(xié)程，那我們是否可以直接用它們來(lái)創(chuàng)建和啟動(dòng)協(xié)程？那當(dāng)然可以。這里我舉個(gè)startCoroutine的例子，仔細(xì)看它的函數(shù)聲明，它其實(shí)是個(gè)擴(kuò)展函數(shù)，擴(kuò)展的是(suspend () -> T)這種類(lèi)型。

(suspend () -> T)：suspend函數(shù)+返回類(lèi)型是T

它可以有2種寫(xiě)法：

//方式1-----------
val block = suspend {
    ...
    "云天明"
}
block.startCoroutine(continuation)
//方式2--------------
suspend fun getUserName(): String {
    ...
    return "云天明"
}
(::getUserName).startCoroutine(continuation)

一種是匿名的suspend函數(shù)，一種是正常的有名字的suspend函數(shù)?，F(xiàn)在，我們簡(jiǎn)單寫(xiě)個(gè)demo來(lái)調(diào)一下startCoroutine。

startCoroutine調(diào)用

//StartCoroutine.kt
fun main() {
    val continuation = object : Continuation<String> {
        override val context: CoroutineContext
            get() = EmptyCoroutineContext
        override fun resumeWith(result: Result<String>) {
            println("結(jié)果: ${result.getOrNull()}")
        }
    }
    block.startCoroutine(continuation)
    Thread.sleep(3000L)
}
val block = suspend {
    println("start")
    delay(2000L)
    println("end")
    "DX3906"
}

調(diào)起非常簡(jiǎn)單，startCoroutine是(suspend () -> T)的擴(kuò)展函數(shù)，且需要傳遞一個(gè)Continuation參數(shù)。我們先反編譯看一下，長(zhǎng)什么樣子。

public final class StartCoroutineKt {
    //block那塊被轉(zhuǎn)換成了一個(gè)類(lèi)StartCoroutineKt$block$1，這里創(chuàng)建好一個(gè)實(shí)例對(duì)象，待會(huì)兒可以直接使用
    private static final Function1<Continuation<? super String>, Object> block = new StartCoroutineKt$block$1((Continuation<? super StartCoroutineKt$block$1>) null);
    public static final void main() {
        //調(diào)用擴(kuò)展函數(shù)，將block和continuation參數(shù)傳入。  
        ContinuationKt.startCoroutine(block, new StartCoroutineKt$main$continuation$1());
        Thread.sleep(3000);
    }
    public static final Function1<Continuation<? super String>, Object> getBlock() {
        return block;
    }
}
//對(duì)應(yīng)block那塊
final class StartCoroutineKt$block$1 extends SuspendLambda implements Function1<Continuation<? super String>, Object> {
    int label;
    StartCoroutineKt$block$1(Continuation<? super StartCoroutineKt$block$1> continuation) {
        super(1, continuation);
    }
    //創(chuàng)建StartCoroutineKt$block$1實(shí)例
    public final Continuation<Unit> create(Continuation<?> continuation) {
        return new StartCoroutineKt$block$1(continuation);
    }
    public final Object invoke(Continuation<? super String> continuation) {
        //創(chuàng)建StartCoroutineKt$block$1實(shí)例并執(zhí)行invokeSuspend
        return ((StartCoroutineKt$block$1) create(continuation)).invokeSuspend(Unit.INSTANCE);
    }
    public final Object invokeSuspend(Object $result) {
        Object coroutine_suspended = IntrinsicsKt.getCOROUTINE_SUSPENDED();
        //狀態(tài)機(jī)
        switch (this.label) {
            case 0:
                //label一開(kāi)始是0
                ResultKt.throwOnFailure($result);
                System.out.println("start");
                this.label = 1;
                //這里正常情況會(huì)返回COROUTINE_SUSPENDED，label已經(jīng)改成1了，下次走case 1的邏輯
                if (DelayKt.delay(2000, this) != coroutine_suspended) {
                    break;
                } else {
                    return coroutine_suspended;
                }
            case 1:
                //label為1，沒(méi)有return，繼續(xù)走最后的結(jié)束語(yǔ)句
                ResultKt.throwOnFailure($result);
                break;
            default:
                throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
        }
        //結(jié)束
        System.out.println("end");
        return "云天明";
    }
}
//對(duì)應(yīng)Continuation那塊
public final class StartCoroutineKt$main$continuation$1 implements Continuation<String> {
    StartCoroutineKt$main$continuation$1() {
    }
    public CoroutineContext getContext() {
        return EmptyCoroutineContext.INSTANCE;
    }
    public void resumeWith(Object result) {
        //輸出結(jié)果
        StringBuilder sb = new StringBuilder();
        sb.append("結(jié)果: ");
        sb.append((String) (Result.m29isFailureimpl(result) ? null : result));
        System.out.println(sb.toString());
    }
}

還是比較清晰的，

首先object : Continuation<String> 是肯定會(huì)生成一個(gè)匿名內(nèi)部類(lèi)，在該類(lèi)中，簡(jiǎn)單在resumeWith里面輸出了一下結(jié)果
block那塊代碼，也會(huì)生成一個(gè)匿名內(nèi)部類(lèi)。需要注意的是，它繼承自SuspendLambda，這個(gè)沒(méi)見(jiàn)過(guò)，待會(huì)兒分析，里面有幾個(gè)方法：create、invoke、invokeSuspend。其中create是創(chuàng)建該類(lèi)的實(shí)例，invoke是調(diào)用create方法并執(zhí)行invokeSuspend，invokeSuspend里面是狀態(tài)機(jī)相關(guān)的邏輯。
main里面執(zhí)行了ContinuationKt.startCoroutine(block, continuation)，調(diào)起了擴(kuò)展方法（擴(kuò)展方法的原理就是這樣的）

反編譯出來(lái)的代碼大致結(jié)構(gòu)我們是了解了，現(xiàn)在需要分析一下startCoroutine具體是怎么走的了，看它是怎么利用這些反編譯出來(lái)的代碼的。

createCoroutineUnintercepted

public fun <T> (suspend () -> T).startCoroutine(
    completion: Continuation<T>
) {
    createCoroutineUnintercepted(completion).intercepted().resume(Unit)
}
//這個(gè)函數(shù)是expect的，沒(méi)有函數(shù)體
public expect fun <T> (suspend () -> T).createCoroutineUnintercepted(
    completion: Continuation<T>
): Continuation<Unit>

startCoroutine首先是調(diào)用了createCoroutineUnintercepted函數(shù)，而createCoroutineUnintercepted是expect的，它是一種聲明。因?yàn)镵otlin是跨平臺(tái)的，所以部分邏輯與平臺(tái)相關(guān)，這個(gè)createCoroutineUnintercepted就是這種。

它沒(méi)有函數(shù)體，我們只關(guān)心JVM平臺(tái)，所以需要到JVM平臺(tái)上找該函數(shù)的實(shí)現(xiàn)。在Kotlin源碼地圖文章中，我們提到協(xié)程源碼，分為2個(gè)倉(cāng)庫(kù)，一個(gè)是Kotlin倉(cāng)庫(kù)，一個(gè)是Kotlin協(xié)程倉(cāng)庫(kù)。

這個(gè)createCoroutineUnintercepted是在Kotlin倉(cāng)庫(kù)中，具體位置是：

kotlin/libraries/stdlib/jvm/src/kotlin/coroutines/intrinsics/IntrinsicsJvm.kt

public actual fun <T> (suspend () -> T).createCoroutineUnintercepted(
    completion: Continuation<T>
): Continuation<Unit> {
    val probeCompletion = probeCoroutineCreated(completion)
    return if (this is BaseContinuationImpl)
        //走這里
        create(probeCompletion)
    else
        createCoroutineFromSuspendFunction(probeCompletion) {
            (this as Function1<Continuation<T>, Any?>).invoke(it)
        }
}

咦，createCoroutineUnintercepted居然也是(suspend () -> T)的擴(kuò)展函數(shù)，所以if那里的this指的就是block，也就是StartCoroutineKt$block$1。它繼承自SuspendLambda。

internal abstract class SuspendLambda(
    public override val arity: Int,
    completion: Continuation<Any?>?
) : ContinuationImpl(completion), FunctionBase<Any?>, SuspendFunction {
    constructor(arity: Int) : this(arity, null)
    public override fun toString(): String =
        if (completion == null)
            Reflection.renderLambdaToString(this) // this is lambda
        else
            super.toString() // this is continuation
}
internal abstract class ContinuationImpl(
    completion: Continuation<Any?>?,
    private val _context: CoroutineContext?
) : BaseContinuationImpl(completion) {
    ......
}
//BaseContinuationImpl實(shí)現(xiàn)了Continuation接口
internal abstract class BaseContinuationImpl(
    public val completion: Continuation<Any?>?
) : Continuation<Any?>, CoroutineStackFrame, Serializable {
    ...
}

SuspendLambda是ContinuationImpl的子類(lèi)，而ContinuationImpl是BaseContinuationImpl的子類(lèi)。所以上面的if (this is BaseContinuationImpl)判斷是ok的，會(huì)走到create(probeCompletion)。

也就是StartCoroutineKt$block$1的create方法，在里面會(huì)創(chuàng)建StartCoroutineKt$block$1實(shí)例。

public final Continuation<Unit> create(Continuation<?> continuation) {
    return new StartCoroutineKt$block$1(continuation);
}

走到這里相當(dāng)于startCoroutine中的createCoroutineUnintercepted(completion)這一步就走完了，它最終返回的是StartCoroutineKt$block$1的實(shí)例，也就是一個(gè)Continuation。它標(biāo)志著協(xié)程被創(chuàng)建好了。再來(lái)看下intercepted是什么邏輯

intercepted

public fun <T> (suspend () -> T).startCoroutine(
    completion: Continuation<T>
) {
    createCoroutineUnintercepted(completion).intercepted().resume(Unit)
}
//好家伙，intercepted也是expect的
public expect fun <T> Continuation<T>.intercepted(): Continuation<T>

發(fā)現(xiàn)這里的intercepted擴(kuò)展函數(shù)也是expect的，又得去kotlin倉(cāng)庫(kù)里面找jvm相關(guān)的實(shí)現(xiàn)。我找了下，路徑在這里：

kotlin/libraries/stdlib/jvm/src/kotlin/coroutines/intrinsics/IntrinsicsJvm.kt

public actual fun <T> Continuation<T>.intercepted(): Continuation<T> =
    (this as? ContinuationImpl)?.intercepted() ?: this

intercepted是一個(gè)擴(kuò)展函數(shù)，這里的this也就是前面createCoroutineUnintercepted(completion)創(chuàng)建出來(lái)的StartCoroutineKt$block$1實(shí)例，它本身是SuspendLambda的子類(lèi)，而SuspendLambda就是ContinuationImpl的子類(lèi)。

所以這里的as?會(huì)轉(zhuǎn)換成功，轉(zhuǎn)換出來(lái)的不是null。也就是說(shuō)走到了ContinuationImpl的intercepted()

internal abstract class ContinuationImpl(
    completion: Continuation<Any?>?,
    private val _context: CoroutineContext?
) : BaseContinuationImpl(completion) {
    constructor(completion: Continuation<Any?>?) : this(completion, completion?.context)
    //這個(gè)context其實(shí)就是傳入的Continuation中的context
    public override val context: CoroutineContext
        get() = _context!!
    @Transient
    private var intercepted: Continuation<Any?>? = null
    public fun intercepted(): Continuation<Any?> =
        intercepted
            ?: (context[ContinuationInterceptor]?.interceptContinuation(this) ?: this)
                .also { intercepted = it }
}
@Transient
private var intercepted: Continuation<Any?>? = null
public fun intercepted(): Continuation<Any?> =
        intercepted
            ?: (context[ContinuationInterceptor]?.interceptContinuation(this) ?: this)
                .also { intercepted = it }

第一次執(zhí)行這里時(shí)intercepted是null，那么會(huì)從context中取ContinuationInterceptor，而context就是Continuation傳入的context，我們傳入的是EmptyCoroutineContext，取出來(lái)是null（ContinuationInterceptor會(huì)對(duì)Continuation進(jìn)行攔截，然后將執(zhí)行邏輯指派到對(duì)應(yīng)的線程之上去，這塊的邏輯后面再細(xì)說(shuō)，就不詳細(xì)展開(kāi)了。）

所以這里intercepted()最終執(zhí)行結(jié)果就是返回this，this也就是StartCoroutineKt$block$1（block函數(shù)生成的類(lèi)）。

intercepted()走完后再回到startCoroutine：

public fun <T> (suspend () -> T).startCoroutine(
    completion: Continuation<T>
) {
    createCoroutineUnintercepted(completion).intercepted().resume(Unit)
}

resume

就差最后一個(gè)resume(Unit)了，前面createCoroutineUnintercepted(completion).intercepted()創(chuàng)建出來(lái)的是StartCoroutineKt$block$1實(shí)例，所以我們需要到這個(gè)類(lèi)里面去找resume函數(shù)。

再提一下類(lèi)的繼承關(guān)系：

StartCoroutineKt$block$1 extends SuspendLambda implements Function1
internal abstract class SuspendLambda(
    public override val arity: Int,
    completion: Continuation<Any?>?
) : ContinuationImpl(completion), FunctionBase<Any?>, SuspendFunction 
internal abstract class ContinuationImpl(
    completion: Continuation<Any?>?,
    private val _context: CoroutineContext?
) : BaseContinuationImpl(completion) 
internal abstract class BaseContinuationImpl(
    public val completion: Continuation<Any?>?
) : Continuation<Any?>, CoroutineStackFrame, Serializable
public interface Continuation<in T> {
    public val context: CoroutineContext
    public fun resumeWith(result: Result<T>)
}

StartCoroutineKt$block$1中沒(méi)有該resume函數(shù)，其父類(lèi)SuspendLambda也沒(méi)有該函數(shù)，再到SuspendLambda的父類(lèi)ContinuationImpl中，發(fā)現(xiàn)也沒(méi)有。再到ContinuationImpl的父類(lèi)BaseContinuationImpl中，也沒(méi)有該函數(shù)，只有一個(gè)resumeWith，奇了怪了。后來(lái)，我發(fā)現(xiàn)這個(gè)resume函數(shù)是一個(gè)擴(kuò)展函數(shù)：

public inline fun <T> Continuation<T>.resume(value: T): Unit =
    resumeWith(Result.success(value))

而resume這個(gè)擴(kuò)展函數(shù)最終是調(diào)用的resumeWith，resumeWidth的實(shí)現(xiàn)在BaseContinuationImpl中。

public final override fun resumeWith(result: Result<Any?>) {
    var current = this
    var param = result
    while (true) {
        probeCoroutineResumed(current)
        with(current) {
            val completion = completion!! // fail fast when trying to resume continuation without completion
            val outcome: Result<Any?> =
                try {
                    val outcome = invokeSuspend(param)
                    if (outcome === COROUTINE_SUSPENDED) return
                    Result.success(outcome)
                } catch (exception: Throwable) {
                    Result.failure(exception)
                }
            releaseIntercepted() // this state machine instance is terminating
            if (completion is BaseContinuationImpl) {
                current = completion
                param = outcome
            } else {
                //label等于1時(shí)走這里
                completion.resumeWith(outcome)
                return
            }
        }
    }
}

這個(gè)開(kāi)了個(gè)while(true)循環(huán)，不斷地執(zhí)行invokeSuspend()，如果遇到invokeSuspend返回結(jié)果是COROUTINE_SUSPENDED則退出while(true)循環(huán)。

public final Object invokeSuspend(Object $result) {
    Object coroutine_suspended = IntrinsicsKt.getCOROUTINE_SUSPENDED();
    //狀態(tài)機(jī)
    switch (this.label) {
        case 0:
            //label一開(kāi)始是0
            ResultKt.throwOnFailure($result);
            System.out.println("start");
            this.label = 1;
            //這里正常情況會(huì)返回COROUTINE_SUSPENDED，label已經(jīng)改成1了，下次走case 1的邏輯
            if (DelayKt.delay(2000, this) != coroutine_suspended) {
                break;
            } else {
                return coroutine_suspended;
            }
        case 1:
            //label為1，沒(méi)有return，繼續(xù)走最后的結(jié)束語(yǔ)句
            ResultKt.throwOnFailure($result);
            break;
        default:
            throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
    }
    //結(jié)束
    System.out.println("end");
    return "云天明";
}

invokeSuspend實(shí)際上就是我們的demo中的StartCoroutineKt$block$1里的invokeSuspend函數(shù)。在demo中，這個(gè)invokeSuspend第一次的時(shí)候狀態(tài)機(jī)那里，label是0，所以會(huì)隨即走到DelayKt.delay(2000, this)，它是一個(gè)掛起函數(shù)，此時(shí)會(huì)拿到結(jié)果：COROUTINE_SUSPENDED。

resumeWith遇到COROUTINE_SUSPENDED就不會(huì)繼續(xù)往下走了，等到delay執(zhí)行完成之后，會(huì)回調(diào)這個(gè)resumeWith函數(shù)，再繼續(xù)走invokeSuspend，此時(shí)label已經(jīng)是1了，走到狀態(tài)機(jī)邏輯那里，返回結(jié)果“云天明”。

這個(gè)結(jié)果會(huì)被resumeWidth的outcome接收住，resumeWidth中的這個(gè)completion其實(shí)就是我們demo中的StartCoroutineKt$main$continuation$1(實(shí)現(xiàn)Continuation<String>的那個(gè)類(lèi)，是通過(guò)構(gòu)造函數(shù)傳進(jìn)來(lái)的)，最終會(huì)走到completion.resumeWith(outcome)，也就是來(lái)到了輸出結(jié)果的地方：println("結(jié)果: ${result.getOrNull()}")。整個(gè)流程就走完了。

結(jié)語(yǔ)

createCoroutine用來(lái)創(chuàng)建協(xié)程，startCoroutine用來(lái)創(chuàng)建并啟動(dòng)協(xié)程。它們內(nèi)部的原理是類(lèi)似的，只是一個(gè)沒(méi)有調(diào)用resume啟動(dòng)協(xié)程，另一個(gè)調(diào)用了resume啟動(dòng)協(xié)程。編譯的時(shí)候，會(huì)生成一個(gè)SuspendLambda的實(shí)現(xiàn)類(lèi)，該類(lèi)invokeSuspend用于執(zhí)行狀態(tài)機(jī)的邏輯，調(diào)用resume后該狀態(tài)機(jī)會(huì)被觸發(fā)，狀態(tài)機(jī)走完，協(xié)程也就走完了。

以上就是Kotlin協(xié)程啟動(dòng)createCoroutine及創(chuàng)建startCoroutine原理的詳細(xì)內(nèi)容，更多關(guān)于Kotlin協(xié)程啟動(dòng)創(chuàng)建的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: