• 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)
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    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)
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13

    一种是匿名的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"
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21

    调起非常简单,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());
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78

    还是比较清晰的,

    • 首先object : Continuation 是肯定会生成一个匿名内部类,在该类中,简单在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>
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

    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)
            }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12

    咦,createCoroutineUnintercepted居然也是(suspend () -> T)的扩展函数,所以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实现了Continuation接口
    internal abstract class BaseContinuationImpl(
        public val completion: Continuation<Any?>?
    ) : Continuation<Any?>, CoroutineStackFrame, Serializable {
        ...
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26

    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);
    }
    
    • 1
    • 2
    • 3

    走到这里相当于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>
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    发现这里的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
    
    • 1
    • 2

    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 }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27

    第一次执行这里时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)
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5

    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>)
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20

    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))
    
    • 1
    • 2

    而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
                }
            }
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27

    这个开了个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 "云天明";
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26

    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的那个类,是通过构造函数传进来的),最终会走到completion.resumeWith(outcome),也就是来到了输出结果的地方:println("结果: ${result.getOrNull()}")。整个流程就走完了。

    二、小结

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

    三、小练习

    大家可以在学习完本篇文章之后,试着去分析一下launch的原理,看看launch与startCoroutine是什么关系。

  • 相关阅读:
    UVM实战——01基本概念_2 什么是UVM?
    Unity求物体关于平面镜像对称后坐标以及旋转
    县域电商数字化转型
    《对比Excel,轻松学习Python数据分析》读书笔记------数据分组与数据透视表
    PHP+茶叶商城系统 毕业设计-附源码211121
    Spring Cloud + Spring Boot + Mybatis + Uniapp 企业架构之 JVM垃圾回收总结
    Rust初探: 实现一个Ping
    Proteus中单片机IO口外接LED输出低电平时,引脚却一直保持高电平的问题(已解决)
    含文档+PPT+源码等]精品基于Nodejs实现的家政服务微信小程序[包运行成功]Nodejs毕业设计计算机项目源码
    Paas 相关介绍
  • 原文地址:https://blog.csdn.net/xfhy_/article/details/126116129