Kotlin协程应用层学习心得

发表于

学习协程的起因

个人了解到协程这个概念是在项目引入kotlin一段时间之后，开始接触到的，但是一直没有深入的学习过这个知识，一直都只是浅尝辄止。一方面难以理解其复杂多样的api，另一方面也没有很强的动力去在生产环境使用协程，毕竟crud仔简单画画页面，写写网络请求，就是这样子了，用不用感觉区别也不大。

后面觉得不得不学习协程有两个契机，一个是学jetpack，里面repository层网络请求一般都使用retrofit，使用的都是协程的写法，配合上livedata/flow,逻辑复杂时，有时候就总感觉没有完全掌握数据的状态变化；另一方面，项目中常常遇到需要同时合并多个异步任务的结果的场景。这是我进行深入学习协程相关使用的原因，至于网上说的消除回调地狱这个问题，我倒感觉没有多重要，毕竟你在回调里再封个方法再调不就没有嵌套了么，所以说这个理由我感觉很牵强。

再说下我学习协程都参考了哪些资料，因为协程的代码一开始对于我来说太难以理解了，同时也走了不少弯路。

一开始我是学习bennyhuo的《深入理解kotlin协程》这本书，以及其相关的视频和他的破解kotlin协程系列文章。这个系列感觉讲的就有点深了，比较少讲到协程的应用，都是协程底层原理和怎么实现一个协程框架，导致在学完之后还是有点晕乎乎的，但是它确实是中文语境下关于协程的一个非常优质的资源。 其中CoroutineLite框架的实现还是常看常新。

后面就是想了解一下协程到底应该怎么使用了，也就是应用层方面的一些东西，这个就是看简中互联网上的一些博客了，包括google的一些官方文章和视频，这些也都讲的很好，但是有时候你能感觉这些知识点都只是简单的堆砌，而且没有连贯起来。

这些应用层的文章里我推荐这一篇，算是里面连贯性比较好，算的上深入浅出的：
segmentfault.com/a/119000004…

后面就看下我个人对于协程的一些体会吧，如果不想看就直接看看我上面说的那些资料吧：）

这里分享的都是协程应用层的一些使用理解，框架层的个人感觉掌握甚浅，也就不在这里班门弄斧了，也欢迎大家多在评论区交流指教。

协程的创建

网上讲协程创建一般会列一大堆api，什么launch，async，runBlocking等等，还会给出一大堆作用域，什么GlobalScope，CoroutineScope，lifecycleScope，viewModelScope等等，搞得大家开头就很懵逼。

其实如果你只做android的话，启动协程的方法，重点学习launch和async就够了，作用域，activity的话使用lifecycleScope，viewModel的话使用viewModelScope，别的一般也用不上。

启动协程，一般这么写就行：

1
2
kotlin复制代码lifecycleScope.launch { ... }
viewModelScope.launch { ... }

async这个api用在我在文章开头说的那种场景，比如你同时发起多个网络请求，并且需要合并请求结果。比如下面这个场景，假设请求的结果都是字符串：

1
2
3
4
5
6
7
8
kotlin复制代码lifecycleScope.launch {
    val deferredA = async { requestA() }
    val deferredB = async { requestB() }
    val deferredC = async { requestC() }
    val sb = StringBuilder()
    sb.append(deferredA.await()).append(deferredB.await())
        .append(deferredC.await())
}

通过这个api可以非常简单的实现这个场景功能，而不必在使用复杂的线程同步框架等工具，至少在这个场景下的异步任务同步化的特性是表现的非常到位的。

另外一个协程的特性我觉得也是其存在的一个必要理由：挂起不阻塞线程，提供线程利用率。

这个特性应该从使用计算机资源的区间维度去理解：

1.进程的出现是为了应对计算机由批处理程序转为并行程序处理场景。

2.线程的出现是为了应对cpu由单核转为多核，需要提高并行程序执行效率的场景。

3.协程的出现是为了应对线程应用开发的复杂场景，将线程执行的动作再微观化为一个个协程的执行块。

比如一个线程同时有三个动作，A是执行一段IO操作，B是执行一段计算，但依赖A的结果，C是执行一段计算，但不依赖A的结果。如果使用线程框架，B和C只能等待A操作完成，这时线程处于阻塞状态；但使用协程框架时，在B等待A执行的同时，C也可以执行。

当然你可以说比如使用类似android的handler机制也可以实现和上面相同的效果，但这就比较复杂了，需要我们定义handler，还需要定制一系列消息，同时进行相关的逻辑控制，使用协程就不需要考虑这么多，还是使用异步任务同步化的写法。后面可以看到协程背面的主线程调度器也是使用handler来完成这个场景的工作的。

launch

1
2
3
4
5
kotlin复制代码public fun CoroutineScope.launch(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> Unit
): Job

launch这个方法的返回值是Job，它可以用线程框架的Thread去理解，可以认为我们获得了一个协程的“句柄”，可以获取到协程此时的生命周期状态，也可以去取消协程，也可以去设置协程的完成和取消回调。

这里看下协程的几个状态：

1
2
3
4
5
6
js复制代码New
Active
Completing
Cancelling
Cancelled
Completed

Job里获取的协程状态反映了这些生命周期：

字段 解释
isActive 活跃的，Job已经开始，还没有完成，取消或失败，则处于active状态。
isComleted 已完成，已取消，已失败和已完成Job都视为完成状态。
isCancelled 已退出，Job由于任何原因取消为true，则视为已退出状态。

async

1
2
3
4
5
kotlin复制代码public fun <T> CoroutineScope.async(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> T
): Deferred<T>

可以看到，api和launch基本一样，只是返回值是Deferred。async的应用场景我们在前面已经说过。

1
2
3
kotlin复制代码public interface Deferred<out T> : Job {
    public suspend fun await(): T
}

Deferred也是一个Job，只是多了await方法而已。

协程的启动

以launch为例进行分析，async的逻辑基本和launch一致。

1
2
3
4
5
6
7
8
9
10
11
12
kotlin复制代码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)
        LazyStandaloneCoroutine(newContext, block) else
        StandaloneCoroutine(newContext, active = true)
    coroutine.start(start, coroutine, block)
    return coroutine
}

前一节看了launch的方法描述，这里继续深入其内部。

1
2
3
4
5
6
7
kotlin复制代码@ExperimentalCoroutinesApi
public actual fun CoroutineScope.newCoroutineContext(context: CoroutineContext): CoroutineContext {
    val combined = coroutineContext + context   //CoroutineContext拼接组合
    val debug = if (DEBUG) combined + CoroutineId(COROUTINE_ID.incrementAndGet()) else combined
    return if (combined !== Dispatchers.Default && combined[ContinuationInterceptor] == null)
        debug + Dispatchers.Default else debug
}

首先是通过newCoroutineContext方法创建了一个协程上下文，这里可以看到，生成的协程上下文既包含我们传入的，也包含CoroutineScope本身的上下文，这个上下文是存在一个“继承”关系的，这也是协程实现结构化并发必不可少的一个要素。

这里组合上下文使用了+，也就是重载运算符方法plus，这里可能再有必要简单介绍下协程上下文。

协程上下文

ok，让我们暂时忘记协程启动这个主线任务，来到协程上下文这个支线任务，我尽量把这个支线任务变得简短，不像网上其他文章那样长篇累牍的描述。

协程上下文可以这么理解：协程中有很多重要的模块，我们需要在协程中获取到这些模块，并驱动其运行对应的功能，这使得需要一个可以在协程中获得这些模块的简单机制，它就是协程上下文。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
kotlin复制代码@SinceKotlin("1.3")
public interface CoroutineContext {
    public operator fun <E : Element> get(key: Key<E>): E?
    public fun <R> fold(initial: R, operation: (R, Element) -> R): R
    public operator fun plus(context: CoroutineContext): CoroutineContext = ...
    public fun minusKey(key: Key<*>): CoroutineContext


    public interface Key<E : Element>


    public interface Element : CoroutineContext {
        public val key: Key<*>
        ...
    }
}

这里我们不对上下文具体实现做过多纠缠，看着上下文接口的api了解下它怎么使用的就行。

协程上下文是一个元素的集合（Element），集合中每个元素唯一，每个元素有一个静态的Key实例，这个数据结构类似一个递归的链表实现，协程中称其为IndexedMap。

get方法使用Key，就可以在集合中找到对应的协程上下文。

plus方法用于在当前协程上下文中加入新的协程上下文，类似链表添加节点。

minuKey方法用于在当前协程上下文中移除某一个协程上下文，类似链表删除节点。

fold用于遍历该容器。

我们已经见过一个协程上下文了，就是launch方法返回的Job，在后面的任务中我们还会遇到更多的协程上下文。

当你迷失在协程代码中时，可以随时拿出来协程上下文，获取到其中你想要的工具，比如：

1
kotlin复制代码coroutineContext[Job]

协程的启动

ok，不是支线任务做不起，而是主线任务更有性价比。让我们回到协程的启动。

1
2
3
4
kotlin复制代码val coroutine = if (start.isLazy)
        LazyStandaloneCoroutine(newContext, block) else
        StandaloneCoroutine(newContext, active = true)
    coroutine.start(start, coroutine, block)

创建协程上下文后，执行这两段代码。

这里可以看到有个分支，分别创建了LazyStandaloneCoroutine和StandaloneCoroutine，先看下它们共同的爸爸：AbstractCoroutine。

1
2
3
4
5
6
7
kotlin复制代码public abstract class AbstractCoroutine<in T>(
parentContext: CoroutineContext,
initParentJob: Boolean,
active: Boolean
) : JobSupport(active), Job, Continuation<T>, CoroutineScope {
    public final override val context: CoroutineContext = parentContext + this
}

AbstractCoroutine真是身兼数职了。

首先它是一个Job，协程上下文，它把自己加了进去。

另外，它也是一个协程作用域，这代表我们启动的协程有一个自己的子作用域。

然后，它也是一个Continuation，协程的运作机制就是靠一个个Continuation衔接起来的，可以将Continuation理解为回调，虽然使用协程api的写法不需要回调，但是协程内部是通过回调来实现功能的，只是在上层api屏蔽了这些东西。

这里的Continuation是一个“completion Continuation”，也就是说，它处于最后一个回调的位置，协程内部所有回调执行完成后，会执行这个结束回调。

Continuation

可以提前看一下Continuation：

1
2
3
4
5
6
7
8
9
10
11
12
kotlin复制代码public interface Continuation<in T> {
    /**
     * The context of the coroutine that corresponds to this continuation.
     */
    public val context: CoroutineContext

    /**
     * Resumes the execution of the corresponding coroutine passing a successful or failed [result] as the
     * return value of the last suspension point.
     */
    public fun resumeWith(result: Result<T>)
}

resumeWith就是Continuation的回调方法。

启动start

1
2
3
4
5
6
7
8
9
10
11
kotlin复制代码public fun <R> start(start: CoroutineStart, receiver: R, block: suspend R.() -> T) {
    start(block, receiver, this)
}

public operator fun <R, T> invoke(block: suspend R.() -> T, receiver: R, completion: Continuation<T>): Unit =
        when (this) {
            DEFAULT -> block.startCoroutineCancellable(receiver, completion)
            ATOMIC -> block.startCoroutine(receiver, completion)
            UNDISPATCHED -> block.startCoroutineUndispatched(receiver, completion)
            LAZY -> Unit // will start lazily
        }

可以看到，协程的启动根据传入的模式走不同的分支，这里只看DEFAULT就行。

1
2
3
4
5
6
7
kotlin复制代码internal fun <R, T> (suspend (R) -> T).startCoroutineCancellable(
    receiver: R, completion: Continuation<T>,
    onCancellation: ((cause: Throwable) -> Unit)? = null
) =
    runSafely(completion) {
        createCoroutineUnintercepted(receiver, completion).intercepted().resumeCancellableWith(Result.success(Unit), onCancellation)
    }

解释下这个方法：

1.根据协程体创建ContinuationImpl。

2.执行拦截器方法，如果有设置调度器，则这里的拦截器指的就是调度器。

3.触发启动协程的执行。

协程的运行

如果直接点击createCoroutineUnintercepted方法，可能看不到具体实现，可以先这么简单理解，后面我们再深入探究：
传入的协程体经过一系列转换，变成了Continuation，它内部使用一套状态机机制去完成协程的运行。

协程拦截器

协程运行的第二步是执行了intercepted()方法：

1
2
3
4
5
6
7
8
kotlin复制代码public actual fun <T> Continuation<T>.intercepted(): Continuation<T> =
    (this as? ContinuationImpl)?.intercepted() ?: this

//ContinuationImpl
public fun intercepted(): Continuation<Any?> =
        intercepted
            ?: (context[ContinuationInterceptor]?.interceptContinuation(this) ?: this)
                .also { intercepted = it }

从协程上下文中获取到了协程拦截器，并执行了它的interceptContinuation方法。

让我们详细了解下协程拦截器，前面也简单提到，它是一个特殊的协程上下文。

1
2
3
4
5
6
kotlin复制代码public interface ContinuationInterceptor : CoroutineContext.Element {
    companion object Key : CoroutineContext.Key<ContinuationInterceptor>
    
    public fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T>
    ...
}

协程拦截器特殊在它总是位于协程上下文集合的最后一位，因为我们需要频繁获取它，所以把它放到最后，降低获取其的时间复杂度。

ok，下面通过一个协程拦截器的示例看下怎么定义一个协程拦截器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
kotlin复制代码private val coroutineInterceptor = object: ContinuationInterceptor {
        override val key: CoroutineContext.Key<*>
            get() = ContinuationInterceptor


        override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> {
            log("init coroutine interceptor")
            return object: Continuation<T> by continuation {


                override fun resumeWith(result: Result<T>) {
                    log("interceptor continuation resumeWith")
                    thread {
                        continuation.resumeWith(result)
                    }
                }
            }
        }
    }

可以看到，interceptContinuation方法基本上就是创建了一个Continuation的代理，并将基本功能委托到原始Continuation，即我们前面所说的ContinuationImpl。

只是重新实现了resumeWith方法，并在其中执行完特殊逻辑后，又转调原始Continuation的resumeWith方法。

协程拦截器就先看到这里，因为实际上基本不需要定义协程拦截器，再简单提两句：

1.如果我们把这个协程拦截器加入到一个复杂的协程中，log("interceptor continuation resumeWith")这个代码执行了很多次，这是和我们前面提到协程的状态机模型机制紧密相关的，这里再卖个关子。

2.在resumeWith方法中，看到是先切了个线程，才执行原Continuation的resumeWith方法，这也是协程调度器即是协程拦截器的本质原因，这个我们马上揭晓。

协程调度器

协程调度器，顾名思义，它可以指定协程运行所在的线程。

1
2
3
4
5
6
7
8
9
10
kotlin复制代码public abstract class CoroutineDispatcher :
    AbstractCoroutineContextElement(ContinuationInterceptor), ContinuationInterceptor {
    ...
        //将可运行块的执行分派到另一个线程。
    public abstract fun dispatch(context: CoroutineContext, block: Runnable)
    ...

     public final override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> =
        DispatchedContinuation(this, continuation)
}

可以看到，协程调度器就是一个协程拦截器，当拦截器运行时，通过dispatch方法将协程调度到指定的线程中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
kotlin复制代码internal class DispatchedContinuation<in T>(
    @JvmField val dispatcher: CoroutineDispatcher,
    @JvmField val continuation: Continuation<T>
) : DispatchedTask<T>(MODE_UNINITIALIZED), CoroutineStackFrame, Continuation<T> by continuation {
    override fun resumeWith(result: Result<T>) {
        val context = continuation.context
        val state = result.toState()
        if (dispatcher.isDispatchNeeded(context)) {
            _state = state
            resumeMode = MODE_ATOMIC
            dispatcher.dispatch(context, this)
        } else {
            executeUnconfined(state, MODE_ATOMIC) {
                withCoroutineContext(this.context, countOrElement) {
                    continuation.resumeWith(result)
                }
            }
        }
    }
}

这个和前面那个协程拦截器的示例很像，可以看到确实执行了dispatch方法。

协程的拦截器声明在Dispatchers.kt中。

1
2
3
kotlin复制代码public actual object Dispatchers {
     public actual val Main: MainCoroutineDispatcher get() = MainDispatcherLoader.dispatcher
}

这里我们看下主线程调度器是怎么实现的，其他的都大同小异。

Android主线程调度器的继承关系，由子到父依次为：HandlerContext – HandlerDispatcher – MainCoroutineDispatcher。

主线程调度器实现了两个关键的方法：

1
2
3
4
5
6
7
8
kotlin复制代码override fun isDispatchNeeded(context: CoroutineContext): Boolean {
        return !invokeImmediately || Looper.myLooper() != handler.looper
}


override fun dispatch(context: CoroutineContext, block: Runnable) {
    handler.post(block)
}

具体执行的逻辑就是，当DispatchedContinuation被触发resumeWith执行时，会先调用isDispatchNeeded，结果发现不在主线程（这里的handler是和主线程looper关联的），就会调用dispatch方法，将逻辑post到主线程执行，这里的逻辑就是被包装ContinuationImpl的resumeWith方法。

由于协程调度器的优先级更高，它会覆盖掉协程声明的拦截器。

协程的运行解密

在前面分析协程启动的过程中，我们又了解到了两个新的协程上下文：协程拦截器和协程调度器。

当把这些元素都准备好之后，终于可以开始执行协程了，来到最后一个方法：

1
kotlin复制代码resumeCancellableWith(Result.success(Unit), onCancellation)

其实就是调用了ContinuationImpl的resumeWith方法。

到这里小总结一下，我们通过launch这个启动协程的api，把启动运行协程的一些基本方法串了一遍，了解到协程大概是这么运行起来的：

首先创建了一个ContinuationImpl，调用了它的resumeWith方法，一通操作后，获取到了协程执行的结果，通过调用Completion Continuation的resumeWith方法，应用层（我们）就获取到了协程运行的结果。

这里可能还是会比较懵，为什么光执行Continuation的resumeWith就能把协程都执行完，我们的协程体代码是怎么样一个存在？

下面就再深入一个层级看下，看完应该都能理解协程运行的大致原理。

ContinuationImpl解密

1
2
3
4
5
6
7
kotlin复制代码internal fun <R, T> (suspend (R) -> T).startCoroutineCancellable(
    receiver: R, completion: Continuation<T>,
    onCancellation: ((cause: Throwable) -> Unit)? = null
) =
    runSafely(completion) {
        createCoroutineUnintercepted(receiver, completion).intercepted().resumeCancellableWith(Result.success(Unit), onCancellation)
    }

这里再搬出来创建协程的这个关键方法，我们这次争取找到createCoroutineUnintercepted的具体实现。
在IntrinsicsJvm.kt这个文件里，终于找到了它的庐山真面目。

1
2
3
4
5
6
7
8
9
10
11
12
13
kotlin复制代码public actual fun <R, T> (suspend R.() -> T).createCoroutineUnintercepted(
    receiver: R,
    completion: Continuation<T>
): Continuation<Unit> {
    val probeCompletion = probeCoroutineCreated(completion)
    return if (this is BaseContinuationImpl)
        create(receiver, probeCompletion)
    else {
        createCoroutineFromSuspendFunction(probeCompletion) {
            (this as Function2<R, Continuation<T>, Any?>).invoke(receiver, it)
        }
    }
}

这里的R指的是协程的作用域，在我们的例子里是lifecycleScope，T指的是我们执行协程的返回值，completion就是创建的StandaloneCoroutine，这个方法的返回值就是ContinuationImpl。 最重要的suspend R.() -> T就是协程体。

先看下else的分支:

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
kotlin复制代码private inline fun <T> createCoroutineFromSuspendFunction(
    completion: Continuation<T>,
    crossinline block: (Continuation<T>) -> Any?
): Continuation<Unit> {
    val context = completion.context
    // label == 0 when coroutine is not started yet (initially) or label == 1 when it was
    return if (context === EmptyCoroutineContext)
        object : RestrictedContinuationImpl(completion as Continuation<Any?>) {
            private var label = 0


            override fun invokeSuspend(result: Result<Any?>): Any? =
                when (label) {
                    0 -> {
                        label = 1
                        result.getOrThrow() // Rethrow exception if trying to start with exception (will be caught by BaseContinuationImpl.resumeWith
                        block(this) // run the block, may return or suspend
                    }
                    1 -> {
                        label = 2
                        result.getOrThrow() // this is the result if the block had suspended
                    }
                    else -> error("This coroutine had already completed")
                }
        }
    else
        object : ContinuationImpl(completion as Continuation<Any?>, context) {
            private var label = 0


            override fun invokeSuspend(result: Result<Any?>): Any? =
                when (label) {
                    0 -> {
                        label = 1
                        result.getOrThrow() // Rethrow exception if trying to start with exception (will be caught by BaseContinuationImpl.resumeWith
                        block(this) // run the block, may return or suspend
                    }
                    1 -> {
                        label = 2
                        result.getOrThrow() // this is the result if the block had suspended
                    }
                    else -> error("This coroutine had already completed")
                }
        }
}

这里重点看else分支里的逻辑，创建了一个ContinuationImpl，并且其invokeSuspend方法执行了传入的协程体。这里向上找下调用invokeSuspend的地方。

ContinuationImpl有个父类BaseContinuationImpl：

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
kotlin复制代码public final override fun resumeWith(result: Result<Any?>) {
        // This loop unrolls recursion in current.resumeWith(param) to make saner and shorter stack traces on resume
        var current = this
        var param = result
        while (true) {
            // Invoke "resume" debug probe on every resumed continuation, so that a debugging library infrastructure
            // can precisely track what part of suspended callstack was already resumed
            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) {
                    // unrolling recursion via loop
                    current = completion
                    param = outcome
                } else {
                    // top-level completion reached -- invoke and return
                    completion.resumeWith(outcome)
                    return
                }
            }
        }

最开始创建出来的ContinuationImpl，执行的是这个resumeWith方法，里面又调用到invokeSuspend，获取到结果后，因为这里的completion（StandaloneCoroutine）不是BaseContinuationImpl，再执行其resumeWith将结果回调通知给应用开发者，即获取到协程体的执行结果。

当然实际情况比这个还要复杂一些，因为创建协程，实际走的是上述if里的逻辑，而不是else里的：

1
2
kotlin复制代码if (this is BaseContinuationImpl)
        create(receiver, probeCompletion)

这是因为我们的协程体被编译器魔改过了，实际的协程体已经变成一个ContinuationImpl的实例，这里的继承关系为：

SuspendLambda：ContinuationImpl：BaseContinuationImpl：Continuation

具体编译器是怎么实现的原理就不再深究了，还是要给发明协程的程序员留口饭吃的（其实是我也不会）。

协程运行的状态机

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
kotlin复制代码public final override fun resumeWith(result: Result<Any?>) {
        // This loop unrolls recursion in current.resumeWith(param) to make saner and shorter stack traces on resume
        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)
                        //返回值为挂起，需要再次调用resumeWith才能继续执行协程体代码
                        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) {
                    // unrolling recursion via loop
                    current = completion
                    param = outcome
                } else {
                    // top-level completion reached -- invoke and return
                    completion.resumeWith(outcome)
                    return
                }
            }
        }
    }

这里再次将BaseContinuationImpl的resumeWith方法摘了出来，这是因为它很关键。

通过前面对协程代码的研究，我们发现了它不仅是协程执行的起点，也是协程执行的发动机。

当协程开始执行时，invokeSuspend被调用，如果该方法返回的是COROUTINE_SUSPENDED，表示协程执行被挂起，这在协程体代码中发反映是调用了一个suspend的挂起函数；

在一次协程执行中，可能会调用多个suspend挂起函数，这时该resumeWith方法也会被执行多次，在这多次执行中，每一次都对应了一个状态机的状态；

直到执行到协程体的结尾，可以获得对应的返回值，这时将获得到的结果用Result封装，并调用completion continuation的resumeWith，将结果回调返回给协程的调用方。

协程运行的状态机示例

这样说可能有些抽象，我们用一个示例来观察一下协程执行，状态机的流转。

首先是协程demo代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
kotlin复制代码fun test() {
        val coroutineDispatcher = newSingleThreadContext("ctx")
        GlobalScope.launch(coroutineDispatcher) {
            println("first coroutine start")
            async(Dispatchers.IO) {
                println("second coroutine start")
                delay(100)
                println("second coroutine end")
            }.await()
            println("first coroutine end")
        }
        Thread.sleep(500)
    }

这是一个简单的协程嵌套，先是通过launch启动了协程1，输出一句日志后，通过async启动了协程2，并通过调用await挂起函数等待返回结果，最后输出了一句日志。

协程2输出了两个日志，并在其中，调用了delay挂起函数延迟了100ms。

下面看下这段协程代码通过编译器生成的状态机示例：

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
kotlin复制代码public final Object invokeSuspend(Object $result) {
    Object var5 = COROUTINE_SUSPEND;
    String var3;
    boolean var4;
    switch(label) {
        case 0:
            ResultKt.throwOnFailure($result);
            CoroutineScope $this$launch = (CoroutineScope)this.L$0;
            var3 = "first coroutine start";
            var4 = false;
            System.out.println(var3);
            Deferred var10000 = BuildersKt.async$default($this$launch, (CoroutineContext)Dispatchers.getIO(), (CoroutineStart)null, new Function2((Continuation)null){
                int label;

                public final Object invokeSuspend(Object $result) {
                    Object var4 = COROUTINE_SUSPEND
                    String var2;
                    boolean var3;
                    switch(label) {
                        case 0:
                            ResultKt.throwOnFailure($result);
                            var2 = "second coroutine start";
                            var3 = false;
                            System.out.println(var2);
                            this.label = 1;
                            if(DelayKt.delay(100L, this) == var4) {
                                return var4;
                            }
                        break;
                        case 1:
                            ResultKt.throwOnFailure($result);
                            break;
                    }
                    var2 = "second coroutine end";
                    var3 = false;        
                    System.out.println(var2);
                    return Unit.INSTANCE;
                }
            })

            ...
            this.label=1;
            if(var10000.await(this) == var5) {
                return var5;
            }
            break;
        case 1:
            break;
    }
    var3 = "first coroutine end";
    var4 = false;
    
}

这个状态机的状态是通过一个int的label变量来维护的，每执行一次挂起函数，这里的label就会加一，并且挂起函数之后的代码就会被分隔到下一段状态机逻辑中。

DE37C017-D332-42CE-A9C6-932CB0F7DC6A.png

对照上图，通过人脑模拟一下这段状态机代码的执行：

1.launch创建协程1，此时创建StandaloneCoroutine（AbstractCoroutine），它同时兼任三项工作。

2.调用CoroutineStart的start方法，即创建并启动协程。

3.启动协程，即是编译器将协程体转化为SuspendLambda（ContinuationImpl，BaseContinuationImpl）的过程。

4.使用intercepted，对Continuation进行代理，此时可能是自定义的协程拦截器，也可能是为了切换线程使用的协程调度器。如果是协程调度器，这里会创建DispatchedContinuation。

5.调用Continuation（BaseContinuationImpl）的resumeWith方法，协程真正开始运行。

6，7.开始执行到invokeSuspend，此时状态机label为0，执行println(“first coroutine start”)，并执行async，创建另一个协程2。

8.状态机label++，变为1，执行到await，这里执行的是DeferredCoroutine的awaitInternal，当前协程被挂起。

9，10，11，12，13，14，15.async协程的执行，这里基本上是和launch上面的过程一致，当调用delay时，又涉及到协程2的挂起和恢复。

16.async协程执行完毕，通过DeferredCoroutine.resumeWith-》makeCompletingOnce-》ResumeAwaitOnCompletion
。这里回调的continuation就是launch协程体，由于此时label已经是1，继续往下执行。
17，18.执行println(“first coroutine end”)，然后协程1也执行完了。

总结

协程其实就是一段可以挂起和恢复执行的运算逻辑，而协程的挂起通过挂起函数实现，挂起函数用状态机的方式用挂起点将协程的运算逻辑拆分成不同的片段，每次运行协程执行不同的逻辑片段。
所以协程有两个很大的好处：

  1. 简化异步编程，支持异步返回；
  2. 挂起不阻塞线程，提供线程利用率。

上面这些内容就是我在学习协程过程中的一些个人的理解，个人觉得最终要的是在协程代码逻辑中，把握住Continuation调用和回调这一关键逻辑，否则很容易陷入人生哲学的三大问题，不知道这时候代码执行到了哪里。

上面这些内容也只是对协程表明的分析，它可以作为继续深入探究协程的前置知识，比如后面重新看bennyhuo的协程分析和CoroutineLite框架，就会更容易一些。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%