前言

在文章正式上干货之前，先说一点背景吧；我是 Kotlin 协程官方文档的译者，大家在 Kotlin 中文官网上看到的绝大多数协程的中文官方文档都是我翻译的。

官方文档可以说是比较全面的介绍了协程的使用，但是就我的感觉来说，这些文档分布的比较散乱，甚至还有三篇分布在协程的官方 Github 的 project 中，很多协程的初学者对这些文档的阅读顺序也尝尝感到摸不到头脑。这里我将一共 15 篇文档的学习顺序做一个整理，如果你还不了解如何使用协程，可以参考我如下的列举：

首先，如果您不了解什么是协程，以及不清楚如何将协程引入您的项目，你可以按顺序阅读这两篇教程：

• 《异步程序设计》
• 《JVM 平台的 Kotlin 协程简介》

接下来两篇官方文档类似于导读或目录，简述了一下协程的理念，以及给出了一些干货的链接，这两篇导读本身倒是没啥干货：

• 《用于异步编程等场景的协程》
• 《协程指南》

然后就是大量的正餐了，如下八篇官方文档介绍了协程使用的方方面面，一定要读懂：

• 《基础》
• 《取消与超时》
• 《通道》
• 《组合挂起函数》
• 《协程上下文与调度器》
• 《异常处理》
• 《Select 表达式》
• 《共享的可变状态与并发》

上面八篇文档读完，再配以大量的实践，你应该已经掌握了协程的基本用法，并开始思考使用它的场景，你可能想知道如何使用协程编写 UI 应用程序，亦或是你可能对协程和响应式流（例如 RxJava）之间的异同和关系有疑问，那么可以参考下面两篇被刊登在官方 Github 上的指南：

• 《使用协程进行 UI 编程指南》
• 《响应式流与协程指南》

现在你应该已经掌握协程在绝大多数场景下的用法，于是你可能好奇于它的实现原理，那么可以阅读这篇官方 Keep：

• 《协程设计文档（KEEP）》

目前这就是协程全部的官方资料，两篇指南和一篇 Keep，都是刊登在 Github 上的；目前 Kotlin 中文站的站长是灰蓝天际老哥，所以上面给出的指向 Github 的地址是指向他 Fork 的版本，以上所有文档的英文原版，都可以在 Kotlin 的英文官网，以及官方的 Github 上找到。

Kotlin 目前是一门多平台语言，虽然协程的设计思想是统一的，但它们在底层的实现原理上会有所不同，例如，在 JVM 和 Android 上，协程的实现要基于线程池的 API，但是在 JS 平台上，由于 JS 本身不支持多线程，所以协程这时必定就不会产生并发。作为一名 Android 工程师，本位将致力于阐述协程在 Android 平台和 JVM 平台的原理，而 JS 平台以及众多的 Native 平台则暂不讨论。

本文将会先介绍一些协程的设计思想，然后详细讲解一下协程的编译相关以及标准库等内容，然后根据源码深入到协程调度器的底层实现细节（调度器这一部分我认为是最值得去看的）。

先从线程说起

协程和线程的关系密不可分，为了能准确的阐述协程的行为，这里有必要先简单描述一下线程是如何执行的。

线程是操作系统的内核资源，是 CPU 调度的最小单位，所有应用程序的代码都运行于线程之上。

无论是回调，还是 RxJava，又或者是 Future 与 Promise，线程都是我们曾经实现并发与异步的最根本的支撑。在 Java 的 API 中，Thread 类是实现线程最基本的类，每创建一个 Thread 对象，就代表着在操作系统内核启动了一个线程，如果我们阅读 Thread 类的源码，可以发现，它的内部实现是大量的 JNI 调用，因为线程的实现必须由操作系统直接提供支持，如果是在 Android 平台上，我们会发现 Thread 的创建过程中，都会调用 Linux API 中的 pthread_create 函数，这直接说明了 Java 层中的 Thread 和 Linux 系统级别的中的线程是一一对应的。

线程的调用存在以下几个问题；首先，线程阻塞与运行两种状态之间的切换有相当大的开销，在传统的线程调用中，线程状态切换的开销一直是程序中一个较大的优化点，例如 Java 在编译时会对锁进行各种优化，例如自旋锁，锁粗化，锁消除等。其次，线程并非是一种轻量级资源，大量创建线程是对系统资源的一种消耗，而传统的阻塞调用会导致系统中存在大量因阻塞而不运行的线程，这对系统资源是一种极大的浪费。

协程与线程不同；首先，协程本质上可以认为是运行在线程上的代码块，协程提供的 挂起 操作会使协程暂停执行，而不会导致线程阻塞。其次，协程是一种轻量级资源，即使创建了上千个协程，对于系统来说也不是一种很大的负担，就如同在 Java 创建上千个 Runable 对象也不会造成过大负担一样。通过这样设计，开发者可以极大的提高线程的使用率，用尽量少的线程执行尽量多的任务，其次调用者无需在编程时思考过多的资源浪费问题，可以在每当有异步或并发需求的时候就不假思索的开启协程。

协程的设计思想

在 Kotlin 中，为了保证安装程序包不会太大（这在 Android 这种嵌入式平台上非常有意义），通常将一些非必须的功能隔离到扩展包中，使用者仅仅在需要时才将它们引入，例如 kotlinx.io（IO）、kotlinx.serialization（序列化）、kotlinx.html（DSL 构建 HTML）、kotlinx.coroutines（协程）等等；这些库我们称之为扩展库，如果我们要使用协程，必须将扩展库引入项目工程，用户直接使用的绝大多数 API 例如：launch、async 等等都由扩展库提供；但是协程从不仅仅是一个库这么简单，它属于 Kotlin 1.3 新增的一种语言特性，所以它的标准库中提供了协程实现的基本原语，扩展库实际上是对这些更底层 API 的封装；除此之外，我们定义挂起函数的“suspend”修饰符属于语言层面的东西，因此需要编译器的直接支持；虽然只有一个“suspend”修饰符，但编译器承担了实现协程的绝大部分任务，可以说是协程的核心，因此在协程的设计思想中，编译器占据了主要的地位，而本节的内容大多数也正是围绕编译器展开。

CPS 变换

CPS 变换是对挂起函数的函数签名进行的一种变换：

我们直接展示一个例子：

挂起函数 await 的函数签名如下所示：

suspend fun <T> CompletableFuture<T>.await(): T

在编译期发生 CPS 变换之后：

fun <T> CompletableFuture<T>.await(continuation: Continuation<T>): Any?

编译器对挂起函数的第一个改变就是对函数签名的改变，这种改变被称为 CPS（续体传递风格）变换。

我们看到发生 CPS 变换后的函数多了一个 Continuation<T> 类型的参数，Continuation 这个单词翻译成中文就是续体，它的声明如下：

interface Continuation<in T> {
   val context: CoroutineContext
   fun resumeWith(result: Result<T>)
}

续体是一个较为抽象的概念，简单来说它包装了协程在挂起之后应该继续执行的代码；在编译的过程中，一个完整的协程被分割切块成一个又一个续体。在 await 函数的挂起结束以后，它会调用 continuation 参数的 resumeWith 函数，来恢复执行 await 函数后面的代码。

最后还要提一点，我们看到发生 CPS 变换的函数，返回值类型变成了 Any?，这是因为这个函数在发生变换后，除了要返回它本身的返回值，还要返回一个标记——COROUTINE_SUSPENDED，而这个返回类型事实上是返回类型 T 与 COROUTINE_SUSPENDED的联合类型，Kotlin 中没有联合类型的概念，貌似也没有加入这种语法的计划，所以只好用最泛化的类型Any?来表示，而COROUTINE_SUSPENDED` 是一个标记，返回它的挂起函数表示这个挂起函数会发生事实上的挂起操作。

续体与续体拦截器

在 CPS 变换 中我们已经了解了续体的基本概念。

挂起函数在恢复的时候，理论上可能会在任何一个线程上恢复，有时我们需要限定协程运行在指定的线程，例如在 UI 编程中，更新 UI 的操作通常只能在 UI 主线程中进行。

我们来看看 Android 的 Main 调度器的实现：

// Main 调度器
@JvmStatic
public actual val Main: MainCoroutineDispatcher get() = MainDispatcherLoader.dispatcher

// dispatcher 由 loadMainDispatcher() 函数创建
internal object MainDispatcherLoader {
    @JvmField
    val dispatcher: MainCoroutineDispatcher = loadMainDispatcher()

    private fun loadMainDispatcher(): MainCoroutineDispatcher {
        ......
    }
}

// MainCoroutineDispatcher
public abstract class MainCoroutineDispatcher : CoroutineDispatcher() {

    @ExperimentalCoroutinesApi
    public abstract val immediate: MainCoroutineDispatcher
}

// CoroutineDispatcher
public abstract class CoroutineDispatcher :
    AbstractCoroutineContextElement(ContinuationInterceptor), ContinuationInterceptor {
    
    ......
}

// ContinuationInterceptor（续体拦截器）
public interface ContinuationInterceptor : CoroutineContext.Element {

    companion object Key : CoroutineContext.Key<ContinuationInterceptor>

    public fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T>

    public fun releaseInterceptedContinuation(continuation: Continuation<*>) {
        /* do nothing by default */
    }

    // Performance optimization for a singleton Key
    public override operator fun <E : CoroutineContext.Element> get(key: CoroutineContext.Key<E>): E? =
        @Suppress("UNCHECKED_CAST")
        if (key === Key) this as E else null

    // Performance optimization to a singleton Key
    public override fun minusKey(key: CoroutineContext.Key<*>): CoroutineContext =
        if (key === Key) EmptyCoroutineContext else this
}

层层追踪让我们找到了 ContinuationInterceptor（续体拦截器），续体拦截器负责拦截恢协程协程在恢复后应执行的代码（即续体）并将其在指定线程或线程池恢复。

在挂起函数的编译中，每个挂起函数都会被编译为一个实现了 Continuation 接口的匿名类，而续体拦截器会拦截真正挂起协程的挂起点的续体。这句话是什么意思？在协程中调用挂起函数，挂起函数不一定会真正挂起协程，例如下面这个例子：

launch {
    val deferred = async {
        // 发起了一个网络请求
        ......
    }
    // 做了一些操作
    ......
    deferred.await()
    // 后续的一些操作
    ......
}

在 deferred.await() 这行执行的时候，如果网络请求已经取得了结果，那 await 函数会直接取得结果，而不会事实上的挂起协程，相反，如果网络请求还未产生结果，await 函数就会使协程挂起。续体拦截器只拦截真正发生挂起的挂起点后的续体，对于未发生挂起的挂起点，续体会被直接调用 resumeWith 这一类的函数而不需要续拦截器对它进行操作。除此之外，续体拦截器还会缓存拦截过的续体，并且在不再需要它的时候调用 releaseInterceptedContinuation 函数释放它。

状态机

协程在编译挂起函数时会将函数体编译为状态机，这样做的好处在于避免创建过多的类和对象，是出于一种性能上的考虑。

我们再来看一个例子，它同样来自 《协程设计文档（KEEP）》：

val a = a()
val y = foo(a).await() // 挂起点 #1
b()
val z = bar(a, y).await() // 挂起点 #2
c(z)

这一段挂起函数内部的代码，它拥有两个挂起点；在编译后生成如下的伪 Java 字节码：

lass <anonymous_for_state_machine> extends SuspendLambda<...> {
    // 状态机当前状态
    int label = 0
    
    // 协程的局部变量
    A a = null
    Y y = null
    
    void resumeWith(Object result) {
        if (label == 0) goto L0
        if (label == 1) goto L1
        if (label == 2) goto L2
        else throw IllegalStateException()
        
      L0:
        // 这次调用，result 应该为空
        a = a()
        label = 1
        result = foo(a).await(this) // 'this' 作为续体传递
        if (result == COROUTINE_SUSPENDED) return // 如果 await 挂起了执行则返回
      L1:
        // 外部代码传入 .await() 的结果恢复协程 
        y = (Y) result
        b()
        label = 2
        result = bar(a, y).await(this) // 'this' 作为续体传递
        if (result == COROUTINE_SUSPENDED) return // 如果 await 挂起了执行则返回
      L2:
        // 外部代码传入 .await() 的结果恢复协程
        Z z = (Z) result
        c(z)
        label = -1 // 没有其他步骤了
        return
    }          
}    

注意，这是伪代码，一个挂起函数会被编译成一个匿名类，这个匿名类中的一个函数实现了这个状态机。我们可以看到，成员变量 label 代表了当前状态机的状态，每一个续体（即挂起点中间的部分以及挂起点与函数头尾之间的部分）都各自对应了一个状态，当函数运行到每个挂起点时，label 的值都受限会发生改变，并且当前的续体（也就是代码中的this）都会作为实参传递给发生了 CPS 变换的挂起函数，如果这个挂起函数没有发生事实上的挂起，函数继续运行，如果发生了事实上的挂起，则函数直接 return。

由于 label 记录了状态，所以，在协程恢复的时候，可以根据状态使用 goto 语句直接跳转至上次的挂起点并向后执行，这就是协程挂起的原理。顺便提一句，虽然 Java 中没有 goto 语句，但是 class 字节码中支持 goto。

CPS 变换 、 续体与续体拦截器 、 状态机 ，这三个小节从细粒度到粗粒度，整体讨论了编译器是如何编译挂起函数的，可见，看似协程只增加了一个 suspend 修饰符，但实际上编译器却做了如此复杂的处理，不得不赞叹协程的设计思想。

标准库

标准库与扩展库的最主要区别就是，扩展库的代码文件的包名通常是“kotlinx.coroutine”，而标准库的包名则是“kotlin.coroutine”。扩展库中的 launch、async 这一类直接提供给开发者使用的 API 我相信不用多说了，我们在项目中使用协程时绝大多数时候都是在和它们打交道，而本节将着重探讨标准库中那些协程操作的最基本的原语。

协程上下文

在前面的小节中我们已经见过了一些标准库中的 API，例如：Continuation、ContinuationInterceptor 等等，除此之外，我们在协程的日常编程中，也见过了一些和扩展库一起使用的 API
例如：CoroutineContext：

interface CoroutineContext {
    operator fun <E : Element> get(key: Key<E>): E?
    fun <R> fold(initial: R, operation: (R, Element) -> R): R
    operator fun plus(context: CoroutineContext): CoroutineContext
    fun minusKey(key: Key<*>): CoroutineContext

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

    interface Key<E : Element>
}

CoroutineContext 是一个接口，所有可以被称为协程上下文的元素都实现了这个接口，这些元素包括了 Dispatchers 以及 Job 等等。

创建以及启动协程

我们在扩展库中通常使用 launch、async 等协程构建器来开启一个协程，但是实际上真正启动协程的函数应该是什么呢？

标准库提供了一个 startCoroutine 函数来启动协程，它是一个扩展函数，接收者是一个挂起函数类型：

@SinceKotlin("1.3")
@Suppress("UNCHECKED_CAST")
public fun <T> (suspend () -> T).startCoroutine(
    completion: Continuation<T>
) {
    createCoroutineUnintercepted(completion).intercepted().resume(Unit)
}

startCoroutine 函数有很多重载，我们选了其中一个来看一下其中的大至实现，我们可以看到，实际上，库会将我们传入的代码块包转成一个续体，然后传入 startCoroutine 函数，所以启动协程仍然是以续体为基本单位的操作。除了 startCoroutine 以外，相关的 API 还有一个 createCoroutine 函数，它与 startCoroutine 的区别在于它只创建一个协程，而不立即执行。

我们从源码中看到 startCoroutine 的内部实现调用了 createCoroutineUnintercepted，它除了有自己的重载以外，还有一个类似的兄弟函数——startCoroutineUninterceptedOrReturn 如果我们去看 launch 或 async 的源码会发现它们两个最后也会调用这个函数，createCoroutineUnintercepted 与 startCoroutineUninterceptedOrReturn 被称为协程内建函数，协程内建函数的概念下一小节会简洁的说一下。

挂起

我们在前面讨论过，调用挂起函数时，协程不一定发生事实上的挂起，那我们如果要编写一个属于自己的可以让协程真正挂起的函数，可以使用哪些 API，例如就像 Java 中提供让线程休眠功能的 Thread.sleep() 或 wait 这些。

根据协程《协程设计文档（KEEP）》的描述，这个低级原语是 suspendCoroutine 函数：

@SinceKotlin("1.3")
@InlineOnly
public suspend inline fun <T> suspendCoroutine(crossinline block: (Continuation<T>) -> Unit): T =
    suspendCoroutineUninterceptedOrReturn { c: Continuation<T> ->
        val safe = SafeContinuation(c.intercepted())
        block(safe)
        safe.getOrThrow()
    }

它调用了 suspendCoroutineUninterceptedOrReturn 函数，如果我们去扒 delay 函数的源码，我们可以发现它最终也调用了 suspendCoroutineUninterceptedOrReturn，按照官方给出的指南，如果你想实现自己的挂起函数，应该使用 suspendCoroutine，它是一个稍微高级一点的封装；至于 suspendCoroutineUninterceptedOrReturn 它如同上一小节的 createCoroutineUnintercepted 一样也是一个协程内建函数：

@SinceKotlin("1.3")
@InlineOnly
@Suppress("UNUSED_PARAMETER", "RedundantSuspendModifier")
public suspend inline fun <T> suspendCoroutineUninterceptedOrReturn(crossinline block: (Continuation<T>) -> Any?): T =
    throw NotImplementedError("Implementation of suspendCoroutineUninterceptedOrReturn is intrinsic")

看到了嘛？它的实现就是直接抛出一个 Error，因为它是一个“黑科技函数”，其实学名叫就叫做协程内建函数，它在 kotlin.coroutines.intrinsics 包下，这个包本身是隐藏的，如果你要使用它，要把它手动 import 到类文件中才能享受 intelliJ IDEA 的自动补全；这种函数的实现没法用 Kotlin 来表达，不过正是因为它“黑科技”，所以编译器其实是认识它们的，会在编译时直接替换它们的实现（我也是第一次见这样的函数）。

但是我们不能仅仅用一个“协程内建函数”的概念就含糊的解释什么是挂起，那挂起的实现到底是怎么实现的，到底谁来调用续体的resum 函数来恢复执行？也就是说必定要有线程一直跟踪这个任务，这样它才能在任务完成时恢复续体的执行。协程的挂起和 Java 的 NIO 机制是类似的，我们在一个线程中执行了一个原本会阻塞线程的任务，但是这个调用者线程没有发生阻塞，这是因为它们有一个专门的线程来负责这些任务的流转，也就是说，当我们发起多个阻塞操作的时候，可能只会阻塞这一个专门的线程，它一直在等待，谁的阻塞结束了，它就把回调再分派过去，这样就完成了阻塞任务与阻塞线程的多对一，而不是以前的一对一，所以挂起也好，NIO 也好，本质上都没有彻底消灭阻塞，但是它们都使阻塞的线程大大减少，从而避免了大量的线程上下文状态切换以及避免了大量线程的产生，从而在 IO 密集型任务中大大提高了性能。

其它

协程库还提供了一些不那么底层的函数，就像扩展库那样的。比如 squence 函数，它是一种协程构建器，用来创建创建一个发射序列的协程，实际作用与扩展库的 Channel 类似。类似的，标准库还提供了可以用来生成一个管道的 buildIterator 协程构建器。squence 与 buildIterator 与扩展库中的那些协程构建器相比最明显的不同就是它们都是顶层函数，即它们不是 CoroutineScope 的扩展函数，在调用时不需要 CoroutineScope 接收者。更多的标准库 API 可以参阅标准库的 API 文档。

实现细节

在了解了设计思想后，协程的大概的实现框架我们就已经了解了，现在我们专注于看两处与 Android 以及 Java 平台相关的平台细节。

Android UI 调度器

我们在介绍续体的时候已经从续体拦截器的角度查看了一部分 Dispatchers.Main 的源码，Dispatchers.Main 事实上是一个多平台化的 API，在 Android、JavaFX、Swing 等场景下实现的细节都不同，我们用 Intelij IDEA 打开 kotlinx.coroutines 项目的源码，这三者的实现都位于 ui 这个目录下。我们在 kotlinx.coroutines.android 包下找到了 HandlerDispatcher 这个代码文件，将协程调度到 Android UI 主线程的具体实现就在其中。

我们回头看一下刚才讲 Dispatchers.Main 这部分的源码，Main 调度器实际上由 MainDispatcherLoader 这个单例的 loadMainDispatcher 函数返回的值。

在 HandlerDispatcher 这个文件中，我们找到了一个密封类：HandlerDispatcher，它继承自 loadMainDispatcher 函数的返回值类型——MainCoroutineDispatcher；
由于是密封类，所以它的所有子类都在这个类文件中，我们找到了它的唯一实现者：HandlerContext：

internal class HandlerContext private constructor(
    private val handler: Handler,
    private val name: String?,
    private val invokeImmediately: Boolean
) : HandlerDispatcher(), Delay {

    public constructor(
        handler: Handler,
        name: String? = null
    ) : this(handler, name, false)

    @Volatile
    private var _immediate: HandlerContext? = if (invokeImmediately) this else null


    override fun isDispatchNeeded(context: CoroutineContext): Boolean {
        return !invokeImmediately || Looper.myLooper() != handler.looper
    }

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

    override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
        val block = Runnable {
            with(continuation) { resumeUndispatched(Unit) }
        }
        handler.postDelayed(block, timeMillis.coerceAtMost(MAX_DELAY))
        continuation.invokeOnCancellation { handler.removeCallbacks(block) }
    }

    override fun invokeOnTimeout(timeMillis: Long, block: Runnable): DisposableHandle {
        handler.postDelayed(block, timeMillis.coerceAtMost(MAX_DELAY))
        return object : DisposableHandle {
            override fun dispose() {
                handler.removeCallbacks(block)
            }
        }
    }
}

HandlerContext 的构造函数的第一个参数就是 Handler，在该文件中，我们找到了调用 HandlerContext 构造函数的地方：

@JvmField
@Deprecated("Use Dispatchers.Main instead", level = DeprecationLevel.HIDDEN)
internal val Main: HandlerDispatcher? = runCatching { HandlerContext(Looper.getMainLooper().asHandler(async = true), "Main") }.getOrNull()

可以看到，传入的 Handler 参数正是 Looper.getMainLooper() 这个 Android 主线程的 Handler。我们再回去看看 HandlerContext 内部的一些函数实现，比如 isDispatchNeeded 这个函数，它表示当前是否需要调度，判断的条件除了 invokeImmediately 这个属性外，另一个条件就是 Looper.myLooper() != handler.looper，用 Looper 来判断当前线程是否是 Handler 对应的线程。除此之外，它调度代码的具体实现是 dispatch 函数，它把你要运行的代码调度到 Android 主线程正是使用了 handler.post，其它几个函数就不多说了，它们也都使用了 Handler 的 API。

所以毫无疑问，在 Android 上，如果要将更新 UI 的代码所在的协程调度到主线程，仍然依赖 Android 的消息机制，并没有任何黑魔法。

Default 以及 IO 调度器的线程池配置

有时候我们可能弄不清楚，我们开启的协程使用 Default 调度器还是 IO 调度器，我们虽然听说过协程最终使用了线程池，但线程池是如何配置的，我们可以从这里去一探究竟：

@JvmStatic
public actual val Default: CoroutineDispatcher = createDefaultDispatcher()
    
......

internal actual fun createDefaultDispatcher(): CoroutineDispatcher =
    if (useCoroutinesScheduler) DefaultScheduler else CommonPool

Default 调度器由 createDefaultDispatcher 函数创建，函数的内部根据 useCoroutinesScheduler 这个成员变量的值进行判断，根据其结果返回 DefaultScheduler 或 CommonPool，useCoroutinesScheduler 的值主要是根据 JVM 的 System.getProperty 方法来获取的，它传入一个包名——“kotlinx.coroutines.scheduler”作为键，然后根据其返回的值是“on”还是“off”来判断 useCoroutinesScheduler 的值是“true”还是“false”，先看看“true”的情况——DefaultScheduler：

internal object DefaultScheduler : ExperimentalCoroutineDispatcher() {
    val IO = blocking(systemProp(IO_PARALLELISM_PROPERTY_NAME, 64.coerceAtLeast(AVAILABLE_PROCESSORS)))

    override fun close() {
        throw UnsupportedOperationException("$DEFAULT_SCHEDULER_NAME cannot be closed")
    }

    override fun toString(): String = DEFAULT_SCHEDULER_NAME

    @InternalCoroutinesApi
    @Suppress("UNUSED")
    public fun toDebugString(): String = super.toString()
}

/**
 * @suppress **This is unstable API and it is subject to change.**
 */
// TODO make internal (and rename) after complete integration
@InternalCoroutinesApi
open class ExperimentalCoroutineDispatcher(
    private val corePoolSize: Int,
    private val maxPoolSize: Int,
    private val idleWorkerKeepAliveNs: Long,
    private val schedulerName: String = "CoroutineScheduler"
) : ExecutorCoroutineDispatcher() {
    constructor(
        corePoolSize: Int = CORE_POOL_SIZE,
        maxPoolSize: Int = MAX_POOL_SIZE,
        schedulerName: String = DEFAULT_SCHEDULER_NAME
    ) : this(corePoolSize, maxPoolSize, IDLE_WORKER_KEEP_ALIVE_NS, schedulerName)

    @Deprecated(message = "Binary compatibility for Ktor 1.0-beta", level = DeprecationLevel.HIDDEN)
    constructor(
        corePoolSize: Int = CORE_POOL_SIZE,
        maxPoolSize: Int = MAX_POOL_SIZE
    ) : this(corePoolSize, maxPoolSize, IDLE_WORKER_KEEP_ALIVE_NS)

    override val executor: Executor
        get() = coroutineScheduler

    // This is variable for test purposes, so that we can reinitialize from clean state
    private var coroutineScheduler = createScheduler()

    override fun dispatch(context: CoroutineContext, block: Runnable): Unit =
        try {
            coroutineScheduler.dispatch(block)
        } catch (e: RejectedExecutionException) {
            DefaultExecutor.dispatch(context, block)
        }

    override fun dispatchYield(context: CoroutineContext, block: Runnable): Unit =
        try {
            coroutineScheduler.dispatch(block, fair = true)
        } catch (e: RejectedExecutionException) {
            DefaultExecutor.dispatchYield(context, block)
        }

    override fun close() = coroutineScheduler.close()

    override fun toString(): String {
        return "${super.toString()}[scheduler = $coroutineScheduler]"
    }

    public fun blocking(parallelism: Int = BLOCKING_DEFAULT_PARALLELISM): CoroutineDispatcher {
        require(parallelism > 0) { "Expected positive parallelism level, but have $parallelism" }
        return LimitingDispatcher(this, parallelism, TaskMode.PROBABLY_BLOCKING)
    }

    public fun limited(parallelism: Int): CoroutineDispatcher {
        require(parallelism > 0) { "Expected positive parallelism level, but have $parallelism" }
        require(parallelism <= corePoolSize) { "Expected parallelism level lesser than core pool size ($corePoolSize), but have $parallelism" }
        return LimitingDispatcher(this, parallelism, TaskMode.NON_BLOCKING)
    }

    internal fun dispatchWithContext(block: Runnable, context: TaskContext, fair: Boolean) {
        try {
            coroutineScheduler.dispatch(block, context, fair)
        } catch (e: RejectedExecutionException) {
            // Context shouldn't be lost here to properly invoke before/after task
            DefaultExecutor.enqueue(coroutineScheduler.createTask(block, context))
        }
    }

    private fun createScheduler() = CoroutineScheduler(corePoolSize, maxPoolSize, idleWorkerKeepAliveNs, schedulerName)

    // fot tests only
    @Synchronized
    internal fun usePrivateScheduler() {
        coroutineScheduler.shutdown(10_000L)
        coroutineScheduler = createScheduler()
    }

    // for tests only
    @Synchronized
    internal fun shutdown(timeout: Long) {
        coroutineScheduler.shutdown(timeout)
    }

    // for tests only
    internal fun restore() = usePrivateScheduler() // recreate scheduler
}

DefaultScheduler 继承自 ExperimentalCoroutineDispatcher，从它的名称和类定义的注释就可以看出来，这是一个非稳定的类，在这个类的内部使用的 require 函数就使用了当前还处于实验性状态的新特性——契约。

从 HandlerDispatcher 中我们可以观察到调度器调度线程的函数是 dispatcher，我们直接去找 ExperimentalCoroutineDispatcher 的 dispatcher 函数，最关键的是这一句：coroutineScheduler.dispatch(block, fair = true)，我们直接去 coroutineScheduler 所属的类去查看 dispatcher 函数：

@Suppress("NOTHING_TO_INLINE")
internal class CoroutineScheduler(
    private val corePoolSize: Int,
    private val maxPoolSize: Int,
    private val idleWorkerKeepAliveNs: Long = IDLE_WORKER_KEEP_ALIVE_NS,
    private val schedulerName: String = DEFAULT_SCHEDULER_NAME
) : Executor, Closeable {
    init {
        require(corePoolSize >= MIN_SUPPORTED_POOL_SIZE) {
            "Core pool size $corePoolSize should be at least $MIN_SUPPORTED_POOL_SIZE"
        }
        require(maxPoolSize >= corePoolSize) {
            "Max pool size $maxPoolSize should be greater than or equals to core pool size $corePoolSize"
        }
        require(maxPoolSize <= MAX_SUPPORTED_POOL_SIZE) {
            "Max pool size $maxPoolSize should not exceed maximal supported number of threads $MAX_SUPPORTED_POOL_SIZE"
        }
        require(idleWorkerKeepAliveNs > 0) {
            "Idle worker keep alive time $idleWorkerKeepAliveNs must be positive"
        }
    }
    
    ······
    
    fun dispatch(block: Runnable, taskContext: TaskContext = NonBlockingContext, fair: Boolean = false) {
        timeSource.trackTask() // this is needed for virtual time support
        val task = createTask(block, taskContext)
        // try to submit the task to the local queue and act depending on the result
        when (submitToLocalQueue(task, fair)) {
            ADDED -> return
            NOT_ADDED -> {
                // try to offload task to global queue
                if (!globalQueue.addLast(task)) {
                    // Global queue is closed in the last step of close/shutdown -- no more tasks should be accepted
                    throw RejectedExecutionException("$schedulerName was terminated")
                }
                requestCpuWorker()
            }
            else -> requestCpuWorker() // ask for help
        }
    }
    
    ······

这个类代码较多，不过我们把大部分都省略了。我们在 Java 中配置线程池最直接的方式是通过构造一个 ThreadPoolExecutor 对象来手动配置，而 ThreadPoolExecutor 类实现了 Executor，在这里，CoroutineScheduler 类也实现了 Executor 接口，可见这就是协程中线程池的一种实现，它没有直接使用 Java 中的 ThreadPoolExecutor API，而是通过实现接口来自己实现了一个线程池。

回到 dispatcher 函数，它将传入的任务压入任务栈，最后通过 requestCpuWorker 函数来执行任务。再继续深入的代码就没必要看了，后面主要就是根据线程池的配置参数，比如 corePoolSize（核心线程数）、maxPoolSize（最大线程数）等来选择是用旧线程执行，还是新启动一个线程执行任务等等。

现在我们看看 corePoolSize 和 corePoolSize 具体是多少。corePoolSize 通过 JVM 的Runtime.getRuntime().availableProcessors() 函数取得当前处理器可运行的线程数，corePoolSize 在大多数情况都等于这个值，它的最小值被设定为 2。而 maxPoolSize 的最小值设定为 corePoolSize，最大值则设定为 (1 shl BLOCKING_SHIFT) - 2，即 1 向左位移 21 位再减 2，此外还要计算一个值 s：Runtime.getRuntime().availableProcessors() 乘以 128。如果 s 介于最小值与最大值之间，则 maxPoolSize 的值为 s，小于最小值等于最小值，大于最大值则等于最大值。

由此可见 DefaultScheduler 这种情况主要用来处理密集型运算，其核心线程数与处理器的线程数相等，这与 RxJava 的计算线程池的思想是类似的。

Default 调度器还有一种情况是 CommonPool；CommonPool 与 DefaultScheduler 的设计是类似的，不过它的线程池没有那么麻烦，它直接使用了 Java 的 Executors.newFixedThreadPool API，这种线程池只有核心线程，核心线程没有超时机制也不会被回收，任务队列没有大小限制，代码就不仔细去看了，总之思想是类似的。

最后看看 IO 调度器：

@JvmStatic
public val IO: CoroutineDispatcher = DefaultScheduler.IO

internal object DefaultScheduler : ExperimentalCoroutineDispatcher() {
    val IO = blocking(systemProp(IO_PARALLELISM_PROPERTY_NAME, 64.coerceAtLeast(AVAILABLE_PROCESSORS)))
    
    public fun blocking(parallelism: Int = BLOCKING_DEFAULT_PARALLELISM): CoroutineDispatcher {
        require(parallelism > 0) { "Expected positive parallelism level, but have $parallelism" }
        return LimitingDispatcher(this, parallelism, TaskMode.PROBABLY_BLOCKING)
    }
    
private class LimitingDispatcher(
    val dispatcher: ExperimentalCoroutineDispatcher,
    val parallelism: Int,
    override val taskMode: TaskMode
) : ExecutorCoroutineDispatcher(), TaskContext, Executor {

override fun dispatch(context: CoroutineContext, block: Runnable) = dispatch(block, false)

    private fun dispatch(block: Runnable, fair: Boolean) {
        var taskToSchedule = block
        while (true) {
            // Commit in-flight tasks slot
            val inFlight = inFlightTasks.incrementAndGet()

            // Fast path, if parallelism limit is not reached, dispatch task and return
            if (inFlight <= parallelism) {
                dispatcher.dispatchWithContext(taskToSchedule, this, fair)
                return
            }

            // Parallelism limit is reached, add task to the queue
            queue.add(taskToSchedule)
            
            if (inFlightTasks.decrementAndGet() >= parallelism) {
                return
            }

            taskToSchedule = queue.poll() ?: return
        }
    }
    
    ......
}

总之就是一层一层向下去找，IO 调度器是 DefaultScheduler 的一个成员属性，最终我们可以找到它的实现类是 LimitingDispatcher——与 CoroutineScheduler 一样，它也是 Executor 接口的实现者，所以它也是一个线程池。我们仔细去阅读它的两个 dispatch 函数，可以发现它的线程调度都依赖自 val dispatcher: ExperimentalCoroutineDispatcher 这个属性的类型正是 DefaultScheduler 的父类，因此，我们知道了，IO 调度器与 Default 调度器的线程池是共用的。

总结一下本小节，IO 调度器与 Default 调度器共用同一个线程池，这个线程池有数量恒定的核心线程，也有数量更大的非核心线程，无论是 IO 任务还是密集型计算任务，都是直接丢向这个线程池，事实上，在协程的使用中，如果使用正确就不存在阻塞的线程，因此我们无需那种池中全是非核心线程的线程池，就像 RxJava 的 IO 调度器一样。这样我们也就得出了一个结论，IO 调度器与 Default 调度器实质上是共用线程的，也就是说我们在使用协程时没有必要特地对 IO 操作使用 IO 调度器。

最后说几句

既然讲解了原理，我们可以总结出协程的量大好处：首先，我们可以用顺序编程的方式实现异步以及并发任务，这比响应式流的链式调用更加直观，给了我们同步编程的体验；其次，在 IO 密集型任务中，协程的挂起操作相比老式的线程阻塞操作大大提高了性能，这成为了协程的杀手级特性之一。最后我来总结一下线程阻塞在协程挂起中的对应特性：

阻塞	挂起
Sychnroized/Lock	Mutex
BlockQueue	Channel
BIO	NIO
sleep	delay
线程安全的容器	暂时无

如果我们要在协程中开启异步任务，请尽量使用上表中“挂起”这一栏下的特性。