Kotlin协程createCoroutine和startCoroutine原理

协程到底是怎么创建和启动的？本篇文章带你揭晓。

createCoroutine 和 startCoroutine

在Continuation.kt文件中，有2个基础API，这里单独提出来说一下，方便后面我们理解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就是用来创建和启动协程的基础API，launch、async等在底层一定程度上都使用了该基础API，launch和async只不过是封装而已。所以，我们先掌握它们。

这2个函数看起来差别不大，一个调用了resume开始了协程，一个没有调用，需要外部去调用resume（createCoroutine会把Continuation返回出去）。

既然launch和async可以用它们来创建和启动协程，那我们是否可以直接用它们来创建和启动协程？那当然可以。这里我举个startCoroutine的例子，仔细看它的函数声明，它其实是个扩展函数，扩展的是(suspend () -> T)这种类型。

(suspend () -> T)：suspend函数+返回类型是T

它可以有2种写法：

//方式1-----------
val block = suspend {
    ...
    "云天明"
}
block.startCoroutine(continuation)

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

一种是匿名的suspend函数，一种是正常的有名字的suspend函数。现在，我们简单写个demo来调一下startCoroutine。

//StartCoroutine.kt
fun main() {
    val continuation = object : Continuation<String> {
        override val context: CoroutineContext
            get() = EmptyCoroutineContext

        override fun resumeWith(result: Result<String>) {
            println("结果: ${result.getOrNull()}")
        }
    }
    block.startCoroutine(continuation)

    Thread.sleep(3000L)
}

val block = suspend {
    println("start")
    delay(2000L)
    println("end")
    "DX3906"
}

调起非常简单，startCoroutine是(suspend () -> T)的扩展函数，且需要传递一个Continuation参数。我们先反编译看一下，长什么样子。

public final class StartCoroutineKt {
    //block那块被转换成了一个类StartCoroutineKt$block$1，这里创建好一个实例对象，待会儿可以直接使用
    private static final Function1<Continuation<? super String>, Object> block = new StartCoroutineKt$block$1((Continuation<? super StartCoroutineKt$block$1>) null);

    public static final void main() {
        //调用扩展函数，将block和continuation参数传入。  
        ContinuationKt.startCoroutine(block, new StartCoroutineKt$main$continuation$1());
        Thread.sleep(3000);
    }

    public static final Function1<Continuation<? super String>, Object> getBlock() {
        return block;
    }
}

//对应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);
    }

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

    public final Object invoke(Continuation<? super String> continuation) {
        //创建StartCoroutineKt$block$1实例并执行invokeSuspend
        return ((StartCoroutineKt$block$1) create(continuation)).invokeSuspend(Unit.INSTANCE);
    }

    public final Object invokeSuspend(Object $result) {
        Object coroutine_suspended = IntrinsicsKt.getCOROUTINE_SUSPENDED();
        //状态机
        switch (this.label) {
            case 0:
                //label一开始是0
                ResultKt.throwOnFailure($result);
                System.out.println("start");
                this.label = 1;
                //这里正常情况会返回COROUTINE_SUSPENDED，label已经改成1了，下次走case 1的逻辑
                if (DelayKt.delay(2000, this) != coroutine_suspended) {
                    break;
                } else {
                    return coroutine_suspended;
                }
            case 1:
                //label为1，没有return，继续走最后的结束语句
                ResultKt.throwOnFailure($result);
                break;
            default:
                throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
        }
        //结束
        System.out.println("end");
        return "云天明";
    }
}

//对应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) {
        //输出结果
        StringBuilder sb = new StringBuilder();
        sb.append("结果: ");
        sb.append((String) (Result.m29isFailureimpl(result) ? null : result));
        System.out.println(sb.toString());
    }
}

还是比较清晰的，

  • 首先object : Continuation<String>是肯定会生成一个匿名内部类，在该类中，简单在resumeWith里面输出了一下结果
  • block那块代码，也会生成一个匿名内部类。需要注意的是，它继承自SuspendLambda，这个没见过，待会儿分析，里面有几个方法：create、invoke、invokeSuspend。其中create是创建该类的实例，invoke是调用create方法并执行invokeSuspend，invokeSuspend里面是状态机相关的逻辑。
  • main里面执行了ContinuationKt.startCoroutine(block, continuation)，调起了扩展方法（扩展方法的原理就是这样的）

反编译出来的代码大致结构我们是了解了，现在需要分析一下startCoroutine具体是怎么走的了，看它是怎么利用这些反编译出来的代码的。

createCoroutineUnintercepted

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

//这个函数是expect的，没有函数体
public expect fun <T> (suspend () -> T).createCoroutineUnintercepted(
    completion: Continuation<T>
): Continuation<Unit>

startCoroutine首先是调用了createCoroutineUnintercepted函数，而createCoroutineUnintercepted是expect的，它是一种声明。因为Kotlin是跨平台的，所以部分逻辑与平台相关，这个createCoroutineUnintercepted就是这种。它没有函数体，我们只关心JVM平台，所以需要到JVM平台上找该函数的实现。在Kotlin源码地图文章中，我们提到协程源码，分为2个仓库，一个是Kotlin仓库，一个是Kotlin协程仓库。这个createCoroutineUnintercepted是在Kotlin仓库中，具体位置是：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)的扩展函数，所以if那里的this指的就是block，也就是StartCoroutineKtblock1。它继承自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实现了Continuation接口
internal abstract class BaseContinuationImpl(
    public val completion: Continuation<Any?>?
) : Continuation<Any?>, CoroutineStackFrame, Serializable {
    ...
}

SuspendLambda是ContinuationImpl的子类，而ContinuationImpl是BaseContinuationImpl的子类。所以上面的if (this is BaseContinuationImpl)判断是ok的，会走到create(probeCompletion)。也就是StartCoroutineKt$block$1的create方法，在里面会创建StartCoroutineKt$block$1实例。

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

走到这里相当于startCoroutine中的createCoroutineUnintercepted(completion)这一步就走完了，它最终返回的是StartCoroutineKt$block$1的实例，也就是一个Continuation。它标志着协程被创建好了。再来看下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>

发现这里的intercepted扩展函数也是expect的，又得去kotlin仓库里面找jvm相关的实现。我找了下，路径在这里：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是一个扩展函数，这里的this也就是前面createCoroutineUnintercepted(completion)创建出来的StartCoroutineKt$block$1实例，它本身是SuspendLambda的子类，而SuspendLambda就是ContinuationImpl的子类。所以这里的as?会转换成功，转换出来的不是null。也就是说走到了ContinuationImpl的intercepted()


internal abstract class ContinuationImpl(
    completion: Continuation<Any?>?,
    private val _context: CoroutineContext?
) : BaseContinuationImpl(completion) {
    constructor(completion: Continuation<Any?>?) : this(completion, completion?.context)
    
    //这个context其实就是传入的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 }

第一次执行这里时intercepted是null，那么会从context中取ContinuationInterceptor，而context就是Continuation传入的context，我们传入的是EmptyCoroutineContext，取出来是null（ContinuationInterceptor会对Continuation进行拦截，然后将执行逻辑指派到对应的线程之上去，这块的逻辑后面再细说，就不详细展开了。），所以这里intercepted()最终执行结果就是返回this，this也就是StartCoroutineKt$block$1（block函数生成的类）。

intercepted()走完后再回到startCoroutine：

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

resume

就差最后一个resume(Unit)了，前面createCoroutineUnintercepted(completion).intercepted()创建出来的是StartCoroutineKt$block$1实例，所以我们需要到这个类里面去找resume函数。

再提一下类的继承关系：

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中没有该resume函数，其父类SuspendLambda也没有该函数，再到SuspendLambda的父类ContinuationImpl中，发现也没有。再到ContinuationImpl的父类BaseContinuationImpl中，也没有该函数，只有一个resumeWith，奇了怪了。后来，我发现这个resume函数是一个扩展函数：

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

而resume这个扩展函数最终是调用的resumeWith，resumeWidth的实现在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时走这里
                completion.resumeWith(outcome)
                return
            }
        }
    }
}

这个开了个while(true)循环，不断地执行invokeSuspend()，如果遇到invokeSuspend返回结果是COROUTINE_SUSPENDED则退出while(true)循环。

public final Object invokeSuspend(Object $result) {
    Object coroutine_suspended = IntrinsicsKt.getCOROUTINE_SUSPENDED();
    //状态机
    switch (this.label) {
        case 0:
            //label一开始是0
            ResultKt.throwOnFailure($result);
            System.out.println("start");
            this.label = 1;
            //这里正常情况会返回COROUTINE_SUSPENDED，label已经改成1了，下次走case 1的逻辑
            if (DelayKt.delay(2000, this) != coroutine_suspended) {
                break;
            } else {
                return coroutine_suspended;
            }
        case 1:
            //label为1，没有return，继续走最后的结束语句
            ResultKt.throwOnFailure($result);
            break;
        default:
            throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
    }
    //结束
    System.out.println("end");
    return "云天明";
}

invokeSuspend实际上就是我们的demo中的StartCoroutineKt$block$1里的invokeSuspend函数。在demo中，这个invokeSuspend第一次的时候状态机那里，label是0，所以会随即走到DelayKt.delay(2000, this)，它是一个挂起函数，此时会拿到结果：COROUTINE_SUSPENDED。resumeWith遇到COROUTINE_SUSPENDED就不会继续往下走了，等到delay执行完成之后，会回调这个resumeWith函数，再继续走invokeSuspend，此时label已经是1了，走到状态机逻辑那里，返回结果“云天明”。

这个结果会被resumeWidth的outcome接收住，resumeWidth中的这个completion其实就是我们demo中的StartCoroutineKt$main$continuation$1(实现Continuation<String>的那个类，是通过构造函数传进来的)，最终会走到completion.resumeWith(outcome)，也就是来到了输出结果的地方：println("结果: ${result.getOrNull()}")。整个流程就走完了。

小结

createCoroutine用来创建协程，startCoroutine用来创建并启动协程。它们内部的原理是类似的，只是一个没有调用resume启动协程，另一个调用了resume启动协程。编译的时候，会生成一个SuspendLambda的实现类，该类invokeSuspend用于执行状态机的逻辑，调用resume后该状态机会被触发，状态机走完，协程也就走完了。

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 203,772评论 6赞 477
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 85,458评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 150,610评论 0赞 337
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,640评论 1赞 276
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,657评论 5赞 365
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,590评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,962评论 3赞 395
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,631评论 0赞 258
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,870评论 1赞 297
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,611评论 2赞 321
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,704评论 1赞 329
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,386评论 4赞 319
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,969评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,944评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,179评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 44,742评论 2赞 349
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,440评论 2赞 342