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开始在下一个可用的免费线程中执行协程(Say Thread01).
3. 在(B),Kotlin停止执行当前线程,并loadData()在下一个可用的空闲线程(Thread02)中启动挂起功能.
4. 当(B)执行后返回时,科特林延续了协程的下一个可用的空闲线程(Thread03).
5. (C)执行Thread03.
6. 在(D),Thread03停止.
7. 在1000ms之后,(E)在下一个免费线程上执行,比方说Thread01.

我理解正确吗？或者协程以不同的方式实现？

Answer 1

协程与您描述的任何调度策略完全不同.协程基本上是suspend funs 的调用链.暂停完全在你的控制之下:你只需要打电话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().

考虑到这一点,让我们重新审视您的图表:

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

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

它可能有也可能没有线程池.UI调度程序使用单个线程.

将线程作为协程调度程序的目标的先决条件是存在与其关联的并发队列,并且该线程运行顶级循环,Runnable该循环从该队列获取对象并执行它们.协程调度程序只是将延续放在该队列上.

2. 在(A),Kotlin开始在下一个可用的免费线程中执行协程(Say Thread01).

它也可以是您调用的同一个线程launch.

3. 在(B),Kotlin停止执行当前线程,并loadData()在下一个可用的空闲线程(Thread02)中启动挂起功能.

Kotlin无需停止任何线程以暂停协程.事实上,协同程序的要点是线程不会启动或停止.线程的顶级循环将继续并选择另一个runnable来运行.

而且,你所谓的一个事实suspend fun并不重要.协程只会在显式调用时才会挂起suspendCoroutine.该功能也可以简单地返回而不会暂停.

但我们假设它确实打过电话 suspendCoroutine.在这种情况下,协程不再在任何线程上运行.它被暂停,直到某个代码在某处调用时才能继续continuation.resume().该代码可以在任何线程上运行,将来的任何时候.

4. 当(B)执行后返回时,科特林延续了协程的下一个可用的空闲线程(Thread03).

B如果没有"执行后返回",协程会在其身体内部恢复时恢复.它可以在返回之前暂停和恢复任意次数.

5. (C)执行Thread03.
6. 在(D),Thread03停止.
7. 在1000ms之后,(E)在下一个免费线程上执行,比方说Thread01.

同样,没有线程被停止.协程被暂停,并且通常专用于调度程序的机制用于在1000毫秒之后安排其恢复.此时,它将被添加到与调度程序关联的运行队列中.

有关特异性,让我们看一些示例,说明发送协程所需的代码类型.

Swing UI调度程序:

EventQueue.invokeLater { continuation.resume(value) }

Android UI调度程序:

mainHandler.post { continuation.resume(value) }

ExecutorService调度程序:

executor.submit { continuation.resume(value) } 

Answer 2

Coroutines work by creating a switch over possible resume points:

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;
    }
}

The resulting code executing this coroutine is then creating that instance and calls the doResume() function everytime it needs to resume execution, how that is handled depends on the scheduler used for execution.

Here is an example compilation for a simple coroutine:

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

Which compiles to this bytecode

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

I compiled this with kotlinc 1.2.41

From 32 to 76 is the code for printing Before and calling delay(1000) which suspends.

From 77 to 101 is the code for printing After.

From 102 to 111 is error handling for illegal resume states, as denoted by the default label in the switch table.

So as a summary, the coroutines in kotlin are simply state-machines that are controlled by some scheduler.