Kotlin协同器是如何内部工作的？

Question

Kotlin是如何在内部实现协同的？

coroutines据说是线程的“较轻版本”，我了解到它们在内部使用线程来执行协同器。

当我使用任何一个构建器函数启动协同线时，会发生什么？

这是我对运行以下代码的理解：

GlobalScope.launch {       <---- (A)
    val y = loadData()     <---- (B)  // suspend fun loadData() 
    println(y)             <---- (C)
    delay(1000)            <---- (D)
    println("completed")   <---- (E)
}

1. Kotlin在开始时有一个预定义的ThreadPool。
2. 在(A)，Kotlin开始在下一个可用的空闲线程(例如Thread01)中执行coroutine。
3. 在(B)，Kotlin停止执行当前线程，并在下一个可用空闲线程(Thread02)中启动挂起函数loadData()。
4. 当(B)在执行后返回时，Kotlin将在下一个可用的空闲线程 (Thread03)中继续使用coroutine 。
5. (C)在Thread03上执行。
6. 在(D)，Thread03被停止。
7. 在1000 is之后，(E)将在下一个空闲线程上执行，比如Thread01。

我理解得对吗？还是协同机制以不同的方式实现？

更新2021年： 这是一篇很棒的文章由曼努埃尔维沃补充以下所有答案。

Answer 1

Coroutines与您描述的任何调度策略都是完全不同的。协同线基本上是suspend fun的一个调用链。暂停完全在您的控制之下:您只需调用suspendCoroutine。您将得到一个回调对象，这样就可以调用它的resume方法并返回到挂起的位置。

下面是一些代码，您可以看到挂起是一种非常直接的、完全由您控制的转换机制：

import kotlin.coroutines.*
import kotlinx.coroutines.*

var continuation: Continuation<String>? = null

fun main(args: Array<String>) {
    val job = GlobalScope.launch(Dispatchers.Unconfined) {
        while (true) {
            println(suspendHere())
        }
    }
    continuation!!.resume("Resumed first time")
    continuation!!.resume("Resumed second time")
}

suspend fun suspendHere() = suspendCancellableCoroutine<String> {
    continuation = it
}

上面的所有代码都在同一个主线程上执行。根本没有多线程正在进行。

您的launch每次调用suspendHere()时都会挂起自己的协同线。它将连续回调写入continuation属性，然后显式地使用该延续来恢复协同工作。

代码使用Unconfined协同调度器，它根本不向线程分派，它只是在调用continuation.resume()的地方运行coroutine代码。

考虑到这一点，让我们重新查看您的图表：

GlobalScope.launch {       <---- (A)
    val y = loadData()     <---- (B)  // suspend fun loadData() 
    println(y)             <---- (C)
    delay(1000)            <---- (D)
    println("completed")   <---- (E)
}

1. Kotlin在开始时有一个预定义的ThreadPool。

它可能有也可能没有线程池。只处理一个线程。

线程成为coroutine的目标的先决条件是，有一个与它相关的并发队列，并且线程运行一个顶级循环，从这个队列中获取Runnable对象并执行它们。coroutine简单地将继续放在该队列上。

1. 在(A)，Kotlin开始在下一个可用的空闲线程(例如Thread01)中执行coroutine。

它也可以是您调用launch的线程。

1. 在(B)，Kotlin停止执行当前线程，并在下一个可用空闲线程(Thread02)中启动挂起函数loadData()。

Kotlin没有必要为了挂起协同线而停止任何线程。事实上，协同工作的要点是线程不会启动或停止。线程的顶级循环将继续，并选择另一个可运行的运行。

此外，仅仅是调用suspend fun这一事实就没有意义了。coroutine只有在显式调用suspendCoroutine时才会挂起自己。函数也可以简单地返回而不暂停。

但让我们假设它确实调用了suspendCoroutine。在这种情况下，协同线不再运行在任何线程上。它被挂起，在某些代码调用continuation.resume()之前不能继续。该代码可以在任何线程上运行，在将来的任何时候都可以运行。

1. 当(B)在执行后返回时，Kotlin将在下一个可用的空闲线程 (Thread03)中继续使用coroutine 。

B不“执行后返回”，协同线恢复时，仍在其体内。在返回之前，它可以挂起并恢复任何次数。

1. (C)在Thread03上执行。
2. 在(D)，Thread03被停止。
3. 在1000 is之后，(E)将在下一个空闲线程上执行，比如Thread01。

同样，没有线程被停止。协同线被挂起，一种机制(通常是特定于调度程序的机制)用于在1000 ms后安排恢复。此时，它将被添加到与dispatcher关联的运行队列中。

为了获得特定性，让我们看一些例子，说明调度协同线所需的代码类型。

Swing UI分配器：

EventQueue.invokeLater { continuation.resume(value) }

Android分配器：

mainHandler.post { continuation.resume(value) }

ExecutorService调度员：

executor.submit { continuation.resume(value) } 

Answer 2

协同工作是通过创建一个可能的简历点来实现的：

class MyClass$Coroutine extends CoroutineImpl {
    public Object doResume(Object o, Throwable t) {
        switch(super.state) {
        default:
                throw new IllegalStateException("call to \"resume\" before \"invoke\" with coroutine");
        case 0:  {
             // code before first suspension
             state = 1; // or something else depending on your branching
             break;
        }
        case 1: {
            ...
        }
        }
        return null;
    }
}

然后，执行这个协同线的生成代码将创建该实例，并在每次需要恢复执行时调用doResume()函数，处理方式取决于用于执行的调度程序。

下面是一个简单协同线的示例编译：

launch {
    println("Before")
    delay(1000)
    println("After")
}

，它编译成这个字节码。

private kotlinx.coroutines.experimental.CoroutineScope p$;

public final java.lang.Object doResume(java.lang.Object, java.lang.Throwable);
Code:
   0: invokestatic  #18                 // Method kotlin/coroutines/experimental/intrinsics/IntrinsicsKt.getCOROUTINE_SUSPENDED:()Ljava/lang/Object;
   3: astore        5
   5: aload_0
   6: getfield      #22                 // Field kotlin/coroutines/experimental/jvm/internal/CoroutineImpl.label:I
   9: tableswitch   { // 0 to 1
                 0: 32
                 1: 77
           default: 102
      }
  32: aload_2
  33: dup
  34: ifnull        38
  37: athrow
  38: pop
  39: aload_0
  40: getfield      #24                 // Field p$:Lkotlinx/coroutines/experimental/CoroutineScope;
  43: astore_3
  44: ldc           #26                 // String Before
  46: astore        4
  48: getstatic     #32                 // Field java/lang/System.out:Ljava/io/PrintStream;
  51: aload         4
  53: invokevirtual #38                 // Method java/io/PrintStream.println:(Ljava/lang/Object;)V
  56: sipush        1000
  59: aload_0
  60: aload_0
  61: iconst_1
  62: putfield      #22                 // Field kotlin/coroutines/experimental/jvm/internal/CoroutineImpl.label:I
  65: invokestatic  #44                 // Method kotlinx/coroutines/experimental/DelayKt.delay:(ILkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
  68: dup
  69: aload         5
  71: if_acmpne     85
  74: aload         5
  76: areturn
  77: aload_2
  78: dup
  79: ifnull        83
  82: athrow
  83: pop
  84: aload_1
  85: pop
  86: ldc           #46                 // String After
  88: astore        4
  90: getstatic     #32                 // Field java/lang/System.out:Ljava/io/PrintStream;
  93: aload         4
  95: invokevirtual #38                 // Method java/io/PrintStream.println:(Ljava/lang/Object;)V
  98: getstatic     #52                 // Field kotlin/Unit.INSTANCE:Lkotlin/Unit;
 101: areturn
 102: new           #54                 // class java/lang/IllegalStateException
 105: dup
 106: ldc           #56                 // String call to \'resume\' before \'invoke\' with coroutine
 108: invokespecial #60                 // Method java/lang/IllegalStateException."<init>":(Ljava/lang/String;)V
 111: athrow

我用kotlinc 1.2.41编译了这个

从32到76是打印Before和调用挂起的delay(1000)的代码。

从77到101是打印After的代码。

从102到111是对非法恢复状态的错误处理，如开关表中的default标签所示。

总之，kotlin中的协同机制只是由一些调度程序控制的状态机。