深入 Kotlin 协程（一）基础设施

Posted 2022-06-17 Updated 2024-05- 15

By Chenhe

24~30 min read

开一个新坑，从底层学起 Kotlin 协程。这意味着本系列文章不会从 kotlinx 那些协程框架入手，直接搞什么结构化并发。相反，从 kotlin 标准库中一般人见不到的函数开始研究，尝试自己实现一套上层协程 API 来强化理解。本系列可看作是《深入理解 Kotlin 协程》@bennyhuo 的心得笔记。

如无特殊说明，本系列中「协程」默认特指 Kotlin 协程的 JVM 实现。

什么是协程

好吧...这真是一个极度刁钻困难的问题。哪怕限定于 Kotlin@JVM 中，也难以回答。困难到两位 GDE bennyhuo 霍丙乾与扔物线朱凯竟给出了矛盾的答案。当然，只是字面描述上的矛盾，思想上还是一致的。

我目前的答案是：协程就是程序自己调度的线程：它有一个函数体作为协程体，由应用程序自己决定何时执行、在哪个线程执行。

当然，这个答案漏掉百出，但我认为至少它揭示或者说隐含了这几个要点：

协程是一段代码，可以被执行、暂停（挂起）、恢复。
协程的调度权在我们自己，而不是操作系统。
所以我们（Kotlin）要自己实现协程调用状态的保存。

一个 Kotlin 协程

创建

使用标准库函数 createCoroutine() 可以创建一个协程，它有一个 Receiver 作为协程体，同时还需要一个 Continuation 类型的参数作为完成回调。(playground)

val coroutine = suspend {
  println("In coroutine")
  "OK" // 协程体的返回值，类型要和完成回调的类型对应。
}.createCoroutine(object : Continuation<String> {
  override val context: CoroutineContext
  get() = EmptyCoroutineContext

  override fun resumeWith(result: Result<String>) {
    println("Coroutine END: $result")
  }
})
coroutine.resume(Unit) // 启动这个协程

// 输出：
// In coroutine
// Coroutine END: Success(OK)

其中比较迷惑的是 Continuation，它不仅是完成回调的类型，同时也是 createCoroutine() 返回的类型。所以...它既是回调又是协程本身？事实上，它只是一个接口，协程本身恰好实现了而已，但不能反过来说它就是协程。

不要把 Continuation 与线程中的 Thread 类比。后者实打实地代表了一个线程，前者只是代表从上次挂起（或从头）开始的一段代码。非要对比的话，Continuation 只能算作线程中的一次「挂起-恢复」。Continuation 是「继续」的意思，也侧面反映了它只是协程中某一段。

线程由操作系统调度，java 无需也无法给我们提供单一「挂起-恢复」的 API。但协程不一样，整个生命周期都是我们自己控制的，所以才出现了这种细粒度的 API。

与 Thread 同级别的东西是 Job，它是 Kotlin 基于协程低级 API 所封装的东西。

协程创建之后需要手动启动。鉴于这个这个操作非常常见，Kotlin 提供了另一个 API startCoroutine() 创建之后顺手启动了，用法和 createCoroutine() 完全一样。

Continuation

刚刚说过，Continuation 只是一个接口，并不是协程本身。跟踪源码发现 createCoroutine() 返回的实际是 SafeContinuation 对象，而这个东西也只是一个包装，一个不到 100 行的迷你类，目的是做一些额外的状态检查，防止外部调用者（也就是我们）不懂协程原理瞎搞，弄出什么奇怪的问题 🐶

作为对比，startCoroutine() 内部就没有做 Safe 包装，因为它不返回给外部，可以看出 Kotlin 的程序猿对自己十分自信 🤓

Safe 是包装，那包装了啥？通过断点调试看到其 delegate 属性是一个类型为 MainKt$main$coroutine$1 的鬼玩意。从字节码的命名看出这是一个匿名内部类。哈？哪来的内部类？这就是协程体（那个 suspend 匿名函数）编译后的结果啦，Kotlin 把它搞成了一个类好能被 JVM 识别。它的字节码如下：

final class MainKt$main$coroutine$1 extends kotlin/coroutines/jvm/internal/SuspendLambda implements kotlin/jvm/functions/Function1

继承了 SuspendLambda，并实现它的抽象函数 invokeSuspend()，把我们协程体编译进去咯。当然，除了我们的协程体之外，还插入了一个 switch 语句，本质是一个状态机，用于实现挂起恢复，不过这是后话了。

至于实现的 Function1 接口，定义如下：

public interface Function1<in P1, out R> : Function<R> {
    /** Invokes the function with the specified argument. */
    public operator fun invoke(p1: P1): R
}

即：这是一个有一个参数和一个返回值的函数。恩？哪来的参数？这就是 suspend 函数的魔法所在，Kotlin 编译器会给它自动加一个类型为 Continuation 的协程实例参数来满足协程的各种需求，后面再仔细研究。

带有 Receiver 的协程体

Kotlin 还有另一组 API 来创建（启动）协程：

public fun <R, T> (suspend R.() -> T).startCoroutine(
    receiver: R,
    completion: Continuation<T>
)

与之前唯一的不同是这个协程体带有 Receiver。但是 Kotlin 并没有声明带有 Receiver 的 Lambda 的语法，下面这种写法是错误的：

// 语法错误
suspend MyScope.{ /*do something*/ }

所以为了愉快地使用这个 API，可以把它封装一下：

fun <R, T> launchCoroutine(receiver: R, block: suspend R.() -> T) {
    block.startCoroutine(receiver, object : Continuation<T> {
        override val context: CoroutineContext
            get() = EmptyCoroutineContext

        override fun resumeWith(result: Result<T>) {
            println("Coroutine END: $result")
        }
    })
}

// 愉快地使用
launchCoroutine(MyScope()){
  println("In coroutine")
}

那么有什么意义？

Receiver 的方法可以在协程体内直接调用了。
可以使用注解 @RestrictsSuspension 限制协程体只能调用 Receiver 里的挂起方法避免意料之外的挂起。特定场景下比较有用。

挂起版 main 函数

从 Kotlin 1.3 开始，可以直接把 main 声明为挂起函数。显然，我们应该认识到这肯定是 Kotlin 编译的花招，因为挂起函数是 Kotlin 的语法，JVM 无法识别这种东西。Kotlin 一定是帮我们生成了一个普通的 main 函数，然后在内部调用挂起版的 main。

看看字节码，证实我们的猜想：

  public static synthetic main([Ljava/lang/String;)V
    NEW MainKt$$$main
    DUP
    ALOAD 0
    INVOKESPECIAL MainKt$$$main.<init> ([Ljava/lang/String;)V
    INVOKESTATIC kotlin/coroutines/jvm/internal/RunSuspendKt.runSuspend (Lkotlin/jvm/functions/Function1;)V
    RETURN

L5 从真正的 main 把启动参数传递给了挂起版本的 main（被编译为一个 Lambda 对象），然后使用 runSuspend 启动它。后者源码如下：

internal fun runSuspend(block: suspend () -> Unit) {
    val run = RunSuspend()
    block.startCoroutine(run)
    run.await()
}

和前面我们自己写的一样，RunSuspend 是协程的完成回调，也在上面的源码里，作用是利用 java 的 wait/notify 机制让 main 等待协程执行完毕，避免程序提前结束。

协程初探

隐含的传参

上面已经提到，协程编译后会自动添加一个类型为协程实例的参数，这也是无参协程体编译后实现了 Function1 接口（拥有一个参数）的原因。为了更形象地理解这一点，可以尝试从 java 侧调用挂起函数。

// Main.kt
suspend fun suspendFunction(): Int {
    // 挂起函数中可通过 suspendCoroutine 拿到当前协程实例 (it)
    return suspendCoroutine { thread { it.resume(5) } }
}

// Test.java
public static void test() {
  Object result = MainKt.suspendFunction(new Continuation<>() {
    @NotNull
    @Override
    public CoroutineContext getContext() {
      return EmptyCoroutineContext.INSTANCE;
    }

    @Override
    public void resumeWith(@NotNull Object o) {
      // coroutine END
    }
  });
}

在 java 中清晰地看到，需要额外传递一个 Continuation 参数，其除了提供协程上下文之外，也承担完成回调的作用，这很像 java 传统的异步写法。那么它的直接返回值是个啥？这要分两种情况讨论：

若函数没有挂起直接返回了结果，那么返回值就是这个结果，可以强转。此时不回调 resumeWith()。
若挂起了，那么返回的就是一个标记（实际是一个枚举值 CoroutineSingletons.COROUTINE_SUSPENDED），表明已经挂起，需要在回调中获取真正的结果。

那为什么回调的类型也是 Object？Kotlin 中应该是 Result<Int>，但是 Result 这个东西比较特殊，好像是 Kotlin 和 java 中间的一个东西，没有对 java 暴露。java 实际接收到的是其包装的值 —— 可能是 int 也可能是 Exception，取公共父类变成 Object 了。

**PS: ** 没错，挂起函数不一定真的挂起，它只是有能力挂起。只有切换了调用栈帧才实打实地挂起。说人话就是，只有在挂起函数返回之后 Continuation.resume() 再被调用才挂起。

协程上下文

协程上下文本质是一个集合类，储存了协程有关的元数据，被保存在协程实例 Continuation 中。用法上类似于 Map，每一个数据都有一个 Key。不同点在于 Key 定义在数据内部，也就是说每一条数据在加入上下文集合之前就已经确定了自己的 Key。

一个典型的上下文 item 类型就是 CoroutineName：

public data class CoroutineName(val name: String) : AbstractCoroutineContextElement(CoroutineName) {
    public companion object Key : CoroutineContext.Key<CoroutineName>
    override fun toString(): String = "CoroutineName($name)"
}

// Key 的定义
public interface Key<E : Element>

不难看出，方便起见，Key 被设计成一个泛型接口，它的实参一般就是 item 的类型。如此一来就不必维护一堆常量用来索引了。假如用 String 作 Key，可以想象，每一个上下文类型都得写一个常量，比如 CoroutineName.KEY = "COROUTINE_NAME" 非常多此一举。

既然上下文是个集合，那么使用自然也和集合类似。在挂起函数中使用 coroutineContext 可获得当前协程的上下文对象，使用 Map 语法取值 (playground)：

suspend {
  println("In coroutine: " + (coroutineContext[CoroutineName] as CoroutineName).name)
}.startCoroutine(object : Continuation<Unit> {
  override val context: CoroutineContext
    get() = CoroutineName("MyCo") // 设置上下文数据

  override fun resumeWith(result: Result<Unit>) {
  }
})

// 输出：
// In coroutine: MyCo

不难得出结论，除了内置的上下文数据类型，我们可以任意定义自己的数据，把它设置给协程并在其中获取、使用。

同样类似 Map，CoroutineContext 重载了 + 运算符，默认实现中相同 Key 的数据会被覆盖。

拦截器

拦截器是实现 AOP 的关键组件，在许多框架中都有它的概念，比如 web 框架中拦截器通常可以一层层过滤 web 请求，再一层层处理响应结果，例如编码成 json 等等。在 Kotlin 协程中，拦截器作用类似，它用于拦截协程的恢复，执行一些操作（例如把它调度到其他线程运行）。

下面我们实现一个简单的拦截器，只打印一下 log：

val logInterceptor = object : ContinuationInterceptor {
  override val key: CoroutineContext.Key<*> = ContinuationInterceptor

  override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> =
    object : Continuation<T> {
      override val context: CoroutineContext = continuation.context

      override fun resumeWith(result: Result<T>) {
        println("Before resume: $result")
        continuation.resumeWith(result)
        println("After resume")
      }
    }
}

拦截器也是协程上下文的一个类型（实现 CoroutineContext.Element），它的核心方法 interceptContinuation() 接收要恢复的协程，把它包装成一个新的实例返回，在这个新协程中可以执行特定的操作发挥拦截器的价值。注意在新的协程中要调用传入协程的 resume 相关方法，否则就成了真正意义上的拦截器了 —— 协程的恢复被拦截，原协程无法继续运行。

既然拦截器也是一个上下文，那么使用方法自然也和上下文一样 (playground)：

suspend {
  println("In coroutine: " + (coroutineContext[CoroutineName] as CoroutineName).name)
}.startCoroutine(object : Continuation<Unit> {
  override val context: CoroutineContext
  	get() = CoroutineName("MyCo") + logInterceptor

  override fun resumeWith(result: Result<Unit>) {
  }
})

// 输出：
// Before resume: Success(kotlin.Unit)
// In coroutine: MyCo
// After resume

协程体中没有调用其他挂起函数，因此只在首次启动时触发一次恢复，之后就一直运行到结束了。

那么拦截器包装函数是怎样执行的？ 它在创建协程的时候就执行了！也就是说协程创建出来，SafeContinuation 里面包裹的不再是协程体，而是经过拦截器包装的 Continuation。查看创建协程的代码，发现它二话不说就调了一下 Continuation.intercepted()，它的 JVM 实现在这里，最终调用的是 ContinuationImpl 的实现：

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

和我们自己写的手动使用协程上下文几乎一样：通过 Key 从 context 中读取拦截器，返回包装后的结果。若没有拦截器就返回自身。

总结

这一篇我们初步探索了协程的实现。估计不少小伙伴看完了之后和没看一样 —— 协程到底怎么用？难道还要手动创建 Continuation 回调？

非也。今天我们使用到的函数都属于 Kotlin 核心库（而不是 kotlinx），它们叫协程的基础设施。之所以那么讲就是因为它们压根就不是给开发者用的。这些 API 足够底层与灵活来用于构建上层 API。如果直接用于生产环境，那无异于用汇编写项目。

写代码的日子

kotlin coroutines

License: 禁止转载到非自托管的内容平台，禁止用于 AI 训练