Kotlin极简教程：第9章 轻量级线程：协程
原文鏈接：https://github.com/EasyKotlin
在常用的并發模型中，多進程、多線程、分布式是最普遍的，不過近些年來逐漸有一些語言以first-class或者library的形式提供對基于協程的并發模型的支持。其中比較典型的有Scheme、Lua、Python、Perl、Go等以first-class的方式提供對協程的支持。
同樣地，Kotlin也支持協程。
本章我們主要介紹：
- 什么是協程
- 協程的用法實例
- 掛起函數
- 通道與管道
- 協程的實現原理
- coroutine庫等
9.1 協程簡介
從硬件發展來看，從最初的單核單CPU，到單核多CPU，多核多CPU，似乎已經到了極限了，但是單核CPU性能卻還在不斷提升。如果將程序分為IO密集型應用和CPU密集型應用，二者的發展歷程大致如下：
IO密集型應用: 多進程->多線程->事件驅動->協程
CPU密集型應用:多進程->多線程
如果說多進程對于多CPU，多線程對應多核CPU，那么事件驅動和協程則是在充分挖掘不斷提高性能的單核CPU的潛力。
常見的有性能瓶頸的API (例如網絡 IO、文件 IO、CPU 或 GPU 密集型任務等)，要求調用者阻塞（blocking）直到它們完成才能進行下一步。后來，我們又使用異步回調的方式來實現非阻塞，但是異步回調代碼寫起來并不簡單。
協程提供了一種避免阻塞線程并用更簡單、更可控的操作替代線程阻塞的方法：協程掛起。
協程主要是讓原來要使用“異步+回調方式”寫出來的復雜代碼, 簡化成可以用看似同步的方式寫出來（對線程的操作進一步抽象）。這樣我們就可以按串行的思維模型去組織原本分散在不同上下文中的代碼邏輯，而不需要去處理復雜的狀態同步問題。
協程最早的描述是由Melvin Conway于1958年給出：“subroutines who act as the master program”(與主程序行為類似的子例程)。此后他又在博士論文中給出了如下定義：
數據在后續調用中始終保持（ The values of data local to a coroutine persist between successive calls 協程的局部）
當控制流程離開時，協程的執行被掛起，此后控制流程再次進入這個協程時，這個協程只應從上次離開掛起的地方繼續 （The execution of a coroutine is suspended as control leaves it, only to carry on where it left off when control re-enters the coroutine at some later stage）。
協程的實現要維護一組局部狀態，在重新進入協程前，保證這些狀態不被改變，從而能順利定位到之前的位置。
協程可以用來解決很多問題，比如nodejs的嵌套回調，Erlang以及Golang的并發模型實現等。
實質上，協程（coroutine）是一種用戶態的輕量級線程。它由協程構建器（launch coroutine builder）啟動。
下面我們通過代碼實踐來學習協程的相關內容。
9.1.1 搭建協程代碼工程
首先，我們來新建一個Kotlin Gradle工程。生成標準gradle工程后，在配置文件build.gradle中，配置kotlinx-coroutines-core依賴：
添加 dependencies :
compile 'org.jetbrains.kotlinx:kotlinx-coroutines-core:0.16'kotlinx-coroutines還提供了下面的模塊：
compile group: 'org.jetbrains.kotlinx', name: 'kotlinx-coroutines-jdk8', version: '0.16' compile group: 'org.jetbrains.kotlinx', name: 'kotlinx-coroutines-nio', version: '0.16' compile group: 'org.jetbrains.kotlinx', name: 'kotlinx-coroutines-reactive', version: '0.16'我們使用Kotlin最新的1.1.3-2 版本:
buildscript {ext.kotlin_version = '1.1.3-2'...dependencies {classpath "org.jetbrains.kotlin:kotlin-gradle-plugin:$kotlin_version"} }其中，kotlin-gradle-plugin是Kotlin集成Gradle的插件。
另外，配置一下JCenter 的倉庫:
repositories {jcenter() }9.1.2 簡單協程示例
下面我們先來看一個簡單的協程示例。
運行下面的代碼：
fun firstCoroutineDemo0() {launch(CommonPool) {delay(3000L, TimeUnit.MILLISECONDS)println("Hello,")}println("World!")Thread.sleep(5000L) }你將會發現輸出：
World! Hello,上面的這段代碼：
launch(CommonPool) {delay(3000L, TimeUnit.MILLISECONDS)println("Hello,") }等價于：
launch(CommonPool, CoroutineStart.DEFAULT, {delay(3000L, TimeUnit.MILLISECONDS)println("Hello, ") })9.1.3 launch函數
這個launch函數定義在kotlinx.coroutines.experimental下面。
public fun launch(context: CoroutineContext,start: CoroutineStart = CoroutineStart.DEFAULT,block: suspend CoroutineScope.() -> Unit ): Job {val newContext = newCoroutineContext(context)val coroutine = if (start.isLazy)LazyStandaloneCoroutine(newContext, block) elseStandaloneCoroutine(newContext, active = true)coroutine.initParentJob(context[Job])start(block, coroutine, coroutine)return coroutine }launch函數有3個入參：context、start、block，這些函數參數分別說明如下：
| context | 協程上下文 |
| start | 協程啟動選項 |
| block | 協程真正要執行的代碼塊，必須是suspend修飾的掛起函數 |
這個launch函數返回一個Job類型，Job是協程創建的后臺任務的概念，它持有該協程的引用。Job接口實際上繼承自CoroutineContext類型。一個Job有如下三種狀態：
| New (optional initial state) 新建 （可選的初始狀態） | false | false |
| Active (default initial state) 活動中（默認初始狀態） | true | false |
| Completed (final state) 已結束（最終狀態） | false | true |
也就是說，launch函數它以非阻塞（non-blocking）當前線程的方式，啟動一個新的協程后臺任務，并返回一個Job類型的對象作為當前協程的引用。
另外，這里的delay()函數類似Thread.sleep()的功能，但更好的是：它不會阻塞線程，而只是掛起協程本身。當協程在等待時，線程將返回到池中, 當等待完成時, 協程將在池中的空閑線程上恢復。
9.1.4 CommonPool：共享線程池
我們再來看一下launch(CommonPool) {...}這段代碼。
首先，這個CommonPool是代表共享線程池，它的主要作用是用來調度計算密集型任務的協程的執行。它的實現使用的是java.util.concurrent包下面的API。它首先嘗試創建一個java.util.concurrent.ForkJoinPool （ForkJoinPool是一個可以執行ForkJoinTask的ExcuteService，它采用了work-stealing模式：所有在池中的線程嘗試去執行其他線程創建的子任務，這樣很少有線程處于空閑狀態，更加高效）；如果不可用，就使用java.util.concurrent.Executors來創建一個普通的線程池：Executors.newFixedThreadPool。相關代碼在kotlinx/coroutines/experimental/CommonPool.kt中：
private fun createPool(): ExecutorService {val fjpClass = Try { Class.forName("java.util.concurrent.ForkJoinPool") }?: return createPlainPool()if (!usePrivatePool) {Try { fjpClass.getMethod("commonPool")?.invoke(null) as? ExecutorService }?.let { return it }}Try { fjpClass.getConstructor(Int::class.java).newInstance(defaultParallelism()) as? ExecutorService }?. let { return it }return createPlainPool() }private fun createPlainPool(): ExecutorService {val threadId = AtomicInteger()return Executors.newFixedThreadPool(defaultParallelism()) {Thread(it, "CommonPool-worker-${threadId.incrementAndGet()}").apply { isDaemon = true }} }這個CommonPool對象類是CoroutineContext的子類型。它們的類型集成層次結構如下：
9.1.5 掛起函數
代碼塊中的delay(3000L, TimeUnit.MILLISECONDS)函數，是一個用suspend關鍵字修飾的函數，我們稱之為掛起函數。掛起函數只能從協程代碼內部調用，普通的非協程的代碼不能調用。
掛起函數只允許由協程或者另外一個掛起函數里面調用, 例如我們在協程代碼中調用一個掛起函數，代碼示例如下：
suspend fun runCoroutineDemo() {run(CommonPool) {delay(3000L, TimeUnit.MILLISECONDS)println("suspend,")}println("runCoroutineDemo!")Thread.sleep(5000L) }fun callSuspendFun() {launch(CommonPool) {runCoroutineDemo()} }如果我們用Java中的Thread類來寫類似功能的代碼，上面的代碼可以寫成這樣：
fun threadDemo0() {Thread({Thread.sleep(3000L)println("Hello,")}).start()println("World!")Thread.sleep(5000L) }輸出結果也是：
World! Hello,另外， 我們不能使用Thread來啟動協程代碼。例如下面的寫法編譯器會報錯：
/*** 錯誤反例：用線程調用協程 error*/ fun threadCoroutineDemo() {Thread({delay(3000L, TimeUnit.MILLISECONDS) // error, Suspend functions are only allowed to be called from a coroutine or another suspend functionprintln("Hello,")})println("World!")Thread.sleep(5000L) }9.2 橋接 阻塞和非阻塞
上面的例子中，我們給出的是使用非阻塞的delay函數，同時又使用了阻塞的Thread.sleep函數，這樣代碼寫在一起可讀性不是那么地好。讓我們來使用純的Kotlin的協程代碼來實現上面的 阻塞+非阻塞 的例子（不用Thread）。
9.2.1 runBlocking函數
Kotlin中提供了runBlocking函數來實現類似主協程的功能：
fun main(args: Array<String>) = runBlocking<Unit> {// 主協程println("${format(Date())}: T0")// 啟動主協程launch(CommonPool) {//在common thread pool中創建協程println("${format(Date())}: T1")delay(3000L)println("${format(Date())}: T2 Hello,")}println("${format(Date())}: T3 World!") // 當子協程被delay，主協程仍然繼續運行delay(5000L)println("${format(Date())}: T4") }運行結果：
14:37:59.640: T0 14:37:59.721: T1 14:37:59.721: T3 World! 14:38:02.763: T2 Hello, 14:38:04.738: T4可以發現，運行結果跟之前的是一樣的，但是我們沒有使用Thread.sleep，我們只使用了非阻塞的delay函數。如果main函數不加 = runBlocking<Unit> , 那么我們是不能在main函數體內調用delay(5000L)的。
如果這個阻塞的線程被中斷，runBlocking拋出InterruptedException異常。
該runBlocking函數不是用來當作普通協程函數使用的，它的設計主要是用來橋接普通阻塞代碼和掛起風格的（suspending style）的非阻塞代碼的, 例如用在 main 函數中，或者用于測試用例代碼中。
@RunWith(JUnit4::class) class RunBlockingTest {@Test fun testRunBlocking() = runBlocking<Unit> {// 這樣我們就可以在這里調用任何suspend fun了launch(CommonPool) {delay(3000L)}delay(5000L)} }9.3 等待一個任務執行完畢
我們先來看一段代碼：
fun firstCoroutineDemo() {launch(CommonPool) {delay(3000L, TimeUnit.MILLISECONDS)println("[firstCoroutineDemo] Hello, 1")}launch(CommonPool, CoroutineStart.DEFAULT, {delay(3000L, TimeUnit.MILLISECONDS)println("[firstCoroutineDemo] Hello, 2")})println("[firstCoroutineDemo] World!") }運行這段代碼，我們會發現只輸出：
[firstCoroutineDemo] World!這是為什么？
為了弄清上面的代碼執行的內部過程，我們打印一些日志看下：
fun testJoinCoroutine() = runBlocking<Unit> {// Start a coroutineval c1 = launch(CommonPool) {println("C1 Thread: ${Thread.currentThread()}")println("C1 Start")delay(3000L)println("C1 World! 1")}val c2 = launch(CommonPool) {println("C2 Thread: ${Thread.currentThread()}")println("C2 Start")delay(5000L)println("C2 World! 2")}println("Main Thread: ${Thread.currentThread()}")println("Hello,")println("Hi,")println("c1 is active: ${c1.isActive} ${c1.isCompleted}")println("c2 is active: ${c2.isActive} ${c2.isCompleted}") }再次運行：
C1 Thread: Thread[ForkJoinPool.commonPool-worker-1,5,main] C1 Start C2 Thread: Thread[ForkJoinPool.commonPool-worker-2,5,main] C2 Start Main Thread: Thread[main,5,main] Hello, Hi, c1 is active: true false c2 is active: true false我們可以看到，這里的C1、C2代碼也開始執行了，使用的是ForkJoinPool.commonPool-worker線程池中的worker線程。但是，我們在代碼執行到最后打印出這兩個協程的狀態isCompleted都是false，這表明我們的C1、C2的代碼，在Main Thread結束的時刻（此時的運行main函數的Java進程也退出了），還沒有執行完畢，然后就跟著主線程一起退出結束了。
所以我們可以得出結論：運行 main () 函數的主線程， 必須要等到我們的協程完成之前結束 , 否則我們的程序在 打印Hello, 1和Hello, 2之前就直接結束掉了。
我們怎樣讓這兩個協程參與到主線程的時間順序里呢？我們可以使用join, 讓主線程一直等到當前協程執行完畢再結束, 例如下面的這段代碼
fun testJoinCoroutine() = runBlocking<Unit> {// Start a coroutineval c1 = launch(CommonPool) {println("C1 Thread: ${Thread.currentThread()}")println("C1 Start")delay(3000L)println("C1 World! 1")}val c2 = launch(CommonPool) {println("C2 Thread: ${Thread.currentThread()}")println("C2 Start")delay(5000L)println("C2 World! 2")}println("Main Thread: ${Thread.currentThread()}")println("Hello,")println("c1 is active: ${c1.isActive} isCompleted: ${c1.isCompleted}")println("c2 is active: ${c2.isActive} isCompleted: ${c2.isCompleted}")c1.join() // the main thread will wait until child coroutine completesprintln("Hi,")println("c1 is active: ${c1.isActive} isCompleted: ${c1.isCompleted}")println("c2 is active: ${c2.isActive} isCompleted: ${c2.isCompleted}")c2.join() // the main thread will wait until child coroutine completesprintln("c1 is active: ${c1.isActive} isCompleted: ${c1.isCompleted}")println("c2 is active: ${c2.isActive} isCompleted: ${c2.isCompleted}") }將會輸出：
C1 Thread: Thread[ForkJoinPool.commonPool-worker-1,5,main] C1 Start C2 Thread: Thread[ForkJoinPool.commonPool-worker-2,5,main] C2 Start Main Thread: Thread[main,5,main] Hello, c1 is active: true isCompleted: false c2 is active: true isCompleted: false C1 World! 1 Hi, c1 is active: false isCompleted: true c2 is active: true isCompleted: false C2 World! 2 c1 is active: false isCompleted: true c2 is active: false isCompleted: true通常，良好的代碼風格我們會把一個單獨的邏輯放到一個獨立的函數中，我們可以重構上面的代碼如下：
fun testJoinCoroutine2() = runBlocking<Unit> {// Start a coroutineval c1 = launch(CommonPool) {fc1()}val c2 = launch(CommonPool) {fc2()}... }private suspend fun fc2() {println("C2 Thread: ${Thread.currentThread()}")println("C2 Start")delay(5000L)println("C2 World! 2") }private suspend fun fc1() {println("C1 Thread: ${Thread.currentThread()}")println("C1 Start")delay(3000L)println("C1 World! 1") }可以看出，我們這里的fc1, fc2函數是suspend fun。
9.4 協程是輕量級的
直接運行下面的代碼：
fun testThread() {val jobs = List(100_1000) {Thread({Thread.sleep(1000L)print(".")})}jobs.forEach { it.start() }jobs.forEach { it.join() } }我們應該會看到輸出報錯：
Exception in thread "main" java.lang.OutOfMemoryError: unable to create new native threadat java.lang.Thread.start0(Native Method)at java.lang.Thread.start(Thread.java:714)at com.easy.kotlin.LightWeightCoroutinesDemo.testThread(LightWeightCoroutinesDemo.kt:30)at com.easy.kotlin.LightWeightCoroutinesDemoKt.main(LightWeightCoroutinesDemo.kt:40) ...........................................................................................我們這里直接啟動了100，000個線程，并join到一起打印”.”, 不出意外的我們收到了java.lang.OutOfMemoryError。
這個異常問題本質原因是我們創建了太多的線程，而能創建的線程數是有限制的，導致了異常的發生。在Java中， 當我們創建一個線程的時候，虛擬機會在JVM內存創建一個Thread對象同時創建一個操作系統線程，而這個系統線程的內存用的不是JVMMemory，而是系統中剩下的內存(MaxProcessMemory - JVMMemory - ReservedOsMemory)。 能創建的線程數的具體計算公式如下：
Number of Threads = (MaxProcessMemory - JVMMemory - ReservedOsMemory) / (ThreadStackSize)
其中，參數說明如下：
| MaxProcessMemory | 指的是一個進程的最大內存 |
| JVMMemory | JVM內存 |
| ReservedOsMemory | 保留的操作系統內存 |
| ThreadStackSize | 線程棧的大小 |
我們通常在優化這種問題的時候，要么是采用減小thread stack的大小的方法，要么是采用減小heap或permgen初始分配的大小方法等方式來臨時解決問題。
在協程中，情況完全就不一樣了。我們看一下實現上面的邏輯的協程代碼：
fun testLightWeightCoroutine() = runBlocking {val jobs = List(100_000) {// create a lot of coroutines and list their jobslaunch(CommonPool) {delay(1000L)print(".")}}jobs.forEach { it.join() } // wait for all jobs to complete }運行上面的代碼，我們將看到輸出：
START: 21:22:28.913 ..................... .....................(100000個) .....END: 21:22:30.956上面的程序在2s左右的時間內正確執行完畢。
9.5 協程 vs 守護線程
在Java中有兩類線程：用戶線程 (User Thread)、守護線程 (Daemon Thread)。
所謂守護線程，是指在程序運行的時候在后臺提供一種通用服務的線程，比如垃圾回收線程就是一個很稱職的守護者，并且這種線程并不屬于程序中不可或缺的部分。因此，當所有的非守護線程結束時，程序也就終止了，同時會殺死進程中的所有守護線程。
我們來看一段Thread的守護線程的代碼：
fun testDaemon2() {val t = Thread({repeat(100) { i ->println("I'm sleeping $i ...")Thread.sleep(500L)}})t.isDaemon = true // 必須在啟動線程前調用,否則會報錯：Exception in thread "main" java.lang.IllegalThreadStateExceptiont.start()Thread.sleep(2000L) // just quit after delay }這段代碼啟動一個線程，并設置為守護線程。線程內部是間隔500ms 重復打印100次輸出。外部主線程睡眠2s。
運行這段代碼，將會輸出：
I'm sleeping 0 ... I'm sleeping 1 ... I'm sleeping 2 ... I'm sleeping 3 ...協程跟守護線程很像，用協程來寫上面的邏輯，代碼如下：
fun testDaemon1() = runBlocking {launch(CommonPool) {repeat(100) { i ->println("I'm sleeping $i ...")delay(500L)}}delay(2000L) // just quit after delay }運行這段代碼，我們發現也輸出：
I'm sleeping 0 ... I'm sleeping 1 ... I'm sleeping 2 ... I'm sleeping 3 ...我們可以看出，活動的協程不會使進程保持活動狀態。它們的行為就像守護程序線程。
9.6 協程執行的取消
我們知道，啟動函數launch返回一個Job引用當前協程，該Job引用可用于取消正在運行協程:
fun testCancellation() = runBlocking<Unit> {val job = launch(CommonPool) {repeat(1000) { i ->println("I'm sleeping $i ... CurrentThread: ${Thread.currentThread()}")delay(500L)}}delay(1300L)println("CurrentThread: ${Thread.currentThread()}")println("Job is alive: ${job.isActive} Job is completed: ${job.isCompleted}")val b1 = job.cancel() // cancels the jobprintln("job cancel: $b1")delay(1300L)println("Job is alive: ${job.isActive} Job is completed: ${job.isCompleted}")val b2 = job.cancel() // cancels the job, job already canceld, return falseprintln("job cancel: $b2")println("main: Now I can quit.") }運行上面的代碼，將會輸出：
I'm sleeping 0 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] I'm sleeping 1 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] I'm sleeping 2 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] CurrentThread: Thread[main,5,main] Job is alive: true Job is completed: false job cancel: true Job is alive: false Job is completed: true job cancel: false main: Now I can quit.我們可以看出，當job還在運行時，isAlive是true，isCompleted是false。當調用job.cancel取消該協程任務，cancel函數本身返回true, 此時協程的打印動作就停止了。此時，job的狀態是isAlive是false，isCompleted是true。 如果，再次調用job.cancel函數，我們將會看到cancel函數返回的是false。
9.6.1 計算代碼的協程取消失效
kotlin 協程的所有suspend 函數都是可以取消的。我們可以通過job的isActive狀態來判斷協程的狀態，或者檢查是否有拋出 CancellationException 時取消。
例如，協程正工作在循環計算中，并且不檢查協程當前的狀態, 那么調用cancel來取消協程將無法停止協程的運行, 如下面的示例所示:
fun testCooperativeCancellation1() = runBlocking<Unit> {val job = launch(CommonPool) {var nextPrintTime = 0Lvar i = 0while (i < 20) { // computation loopval currentTime = System.currentTimeMillis()if (currentTime >= nextPrintTime) {println("I'm sleeping ${i++} ... CurrentThread: ${Thread.currentThread()}")nextPrintTime = currentTime + 500L}}}delay(3000L)println("CurrentThread: ${Thread.currentThread()}")println("Before cancel, Job is alive: ${job.isActive} Job is completed: ${job.isCompleted}")val b1 = job.cancel() // cancels the jobprintln("job cancel1: $b1")println("After Cancel, Job is alive: ${job.isActive} Job is completed: ${job.isCompleted}")delay(30000L)val b2 = job.cancel() // cancels the job, job already canceld, return falseprintln("job cancel2: $b2")println("main: Now I can quit.") }運行上面的代碼，輸出：
I'm sleeping 0 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] I'm sleeping 1 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] ... I'm sleeping 6 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] CurrentThread: Thread[main,5,main] Before cancel, Job is alive: true Job is completed: false job cancel1: true After Cancel, Job is alive: false Job is completed: true I'm sleeping 7 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] ... I'm sleeping 18 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] I'm sleeping 19 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] job cancel2: false main: Now I can quit.我們可以看出，即使我們調用了cancel函數，當前的job狀態isAlive是false了，但是協程的代碼依然一直在運行，并沒有停止。
9.6.2 計算代碼協程的有效取消
有兩種方法可以使計算代碼取消成功。
方法一： 顯式檢查取消狀態isActive
我們直接給出實現的代碼：
fun testCooperativeCancellation2() = runBlocking<Unit> {val job = launch(CommonPool) {var nextPrintTime = 0Lvar i = 0while (i < 20) { // computation loopif (!isActive) {return@launch}val currentTime = System.currentTimeMillis()if (currentTime >= nextPrintTime) {println("I'm sleeping ${i++} ... CurrentThread: ${Thread.currentThread()}")nextPrintTime = currentTime + 500L}}}delay(3000L)println("CurrentThread: ${Thread.currentThread()}")println("Before cancel, Job is alive: ${job.isActive} Job is completed: ${job.isCompleted}")val b1 = job.cancel() // cancels the jobprintln("job cancel1: $b1")println("After Cancel, Job is alive: ${job.isActive} Job is completed: ${job.isCompleted}")delay(3000L)val b2 = job.cancel() // cancels the job, job already canceld, return falseprintln("job cancel2: $b2")println("main: Now I can quit.") }運行這段代碼，輸出：
I'm sleeping 0 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] I'm sleeping 1 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] I'm sleeping 2 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] I'm sleeping 3 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] I'm sleeping 4 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] I'm sleeping 5 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] I'm sleeping 6 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] CurrentThread: Thread[main,5,main] Before cancel, Job is alive: true Job is completed: false job cancel1: true After Cancel, Job is alive: false Job is completed: true job cancel2: false main: Now I can quit.正如您所看到的, 現在這個循環可以被取消了。這里的isActive屬性是CoroutineScope中的屬性。這個接口的定義是：
public interface CoroutineScope {public val isActive: Booleanpublic val context: CoroutineContext }該接口用于通用協程構建器的接收器，以便協程中的代碼可以方便的訪問其isActive狀態值（取消狀態），以及其上下文CoroutineContext信息。
方法二： 循環調用一個掛起函數yield()
該方法實質上是通過job的isCompleted狀態值來捕獲CancellationException完成取消功能。
我們只需要在while循環體中循環調用yield()來檢查該job的取消狀態，如果已經被取消，那么isCompleted值將會是true，yield函數就直接拋出CancellationException異常，從而完成取消的功能：
val job = launch(CommonPool) {var nextPrintTime = 0Lvar i = 0while (i < 20) { // computation loopyield()val currentTime = System.currentTimeMillis()if (currentTime >= nextPrintTime) {println("I'm sleeping ${i++} ... CurrentThread: ${Thread.currentThread()}")nextPrintTime = currentTime + 500L}} }運行上面的代碼，輸出：
I'm sleeping 0 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-1,5,main] I'm sleeping 1 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-2,5,main] I'm sleeping 2 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-2,5,main] I'm sleeping 3 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-3,5,main] I'm sleeping 4 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-3,5,main] I'm sleeping 5 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-3,5,main] I'm sleeping 6 ... CurrentThread: Thread[ForkJoinPool.commonPool-worker-2,5,main] CurrentThread: Thread[main,5,main] Before cancel, Job is alive: true Job is completed: false job cancel1: true After Cancel, Job is alive: false Job is completed: true job cancel2: false main: Now I can quit.如果我們想看看yield函數拋出的異常，我們可以加上try catch打印出日志：
try {yield() } catch (e: Exception) {println("$i ${e.message}") }我們可以看到類似：Job was cancelled 這樣的信息。
這個yield函數的實現是：
suspend fun yield(): Unit = suspendCoroutineOrReturn sc@ { cont ->val context = cont.contextval job = context[Job]if (job != null && job.isCompleted) throw job.getCompletionException()if (cont !is DispatchedContinuation<Unit>) return@sc Unitif (!cont.dispatcher.isDispatchNeeded(context)) return@sc Unitcont.dispatchYield(job, Unit)COROUTINE_SUSPENDED }如果調用此掛起函數時，當前協程的Job已經完成 (isActive = false, isCompleted = true)，當前協程將以CancellationException取消。
9.6.3 在finally中的協程代碼
當我們取消一個協程任務時，如果有try {...} finally {...}代碼塊，那么finally {…}中的代碼會被正常執行完畢：
fun finallyCancelDemo() = runBlocking {val job = launch(CommonPool) {try {repeat(1000) { i ->println("I'm sleeping $i ...")delay(500L)}} finally {println("I'm running finally")}}delay(2000L)println("Before cancel, Job is alive: ${job.isActive} Job is completed: ${job.isCompleted}")job.cancel()println("After cancel, Job is alive: ${job.isActive} Job is completed: ${job.isCompleted}")delay(2000L)println("main: Now I can quit.") }運行這段代碼，輸出：
I'm sleeping 0 ... I'm sleeping 1 ... I'm sleeping 2 ... I'm sleeping 3 ... Before cancel, Job is alive: true Job is completed: false I'm running finally After cancel, Job is alive: false Job is completed: true main: Now I can quit.我們可以看出，在調用cancel之后，就算當前協程任務Job已經結束了，finally{...}中的代碼依然被正常執行。
但是，如果我們在finally{...}中放入掛起函數：
fun finallyCancelDemo() = runBlocking {val job = launch(CommonPool) {try {repeat(1000) { i ->println("I'm sleeping $i ...")delay(500L)}} finally {println("I'm running finally")delay(1000L)println("And I've delayed for 1 sec ?")}}delay(2000L)println("Before cancel, Job is alive: ${job.isActive} Job is completed: ${job.isCompleted}")job.cancel()println("After cancel, Job is alive: ${job.isActive} Job is completed: ${job.isCompleted}")delay(2000L)println("main: Now I can quit.") }運行上述代碼，我們將會發現只輸出了一句：I’m running finally。因為主線程在掛起函數delay(1000L)以及后面的打印邏輯還沒執行完，就已經結束退出。
finally {println("I'm running finally")delay(1000L)println("And I've delayed for 1 sec ?") }9.6.4 協程執行不可取消的代碼塊
如果我們想要上面的例子中的finally{...}完整執行，不被取消函數操作所影響，我們可以使用 run 函數和 NonCancellable 上下文將相應的代碼包裝在 run (NonCancellable) {…} 中, 如下面的示例所示:
fun testNonCancellable() = runBlocking {val job = launch(CommonPool) {try {repeat(1000) { i ->println("I'm sleeping $i ...")delay(500L)}} finally {run(NonCancellable) {println("I'm running finally")delay(1000L)println("And I've just delayed for 1 sec because I'm non-cancellable")}}}delay(2000L)println("main: I'm tired of waiting!")job.cancel()delay(2000L)println("main: Now I can quit.") }運行輸出：
I'm sleeping 0 ... I'm sleeping 1 ... I'm sleeping 2 ... I'm sleeping 3 ... main: I'm tired of waiting! I'm running finally And I've just delayed for 1 sec because I'm non-cancellable main: Now I can quit.9.7 設置協程超時時間
我們通常取消協同執行的原因給協程的執行時間設定一個執行時間上限。我們也可以使用 withTimeout 函數來給一個協程任務的執行設定最大執行時間，超出這個時間，就直接終止掉。代碼示例如下:
fun testTimeouts() = runBlocking {withTimeout(3000L) {repeat(100) { i ->println("I'm sleeping $i ...")delay(500L)}} }運行上述代碼，我們將會看到如下輸出：
I'm sleeping 0 ... I'm sleeping 1 ... I'm sleeping 2 ... I'm sleeping 3 ... I'm sleeping 4 ... I'm sleeping 5 ... Exception in thread "main" kotlinx.coroutines.experimental.TimeoutException: Timed out waiting for 3000 MILLISECONDSat kotlinx.coroutines.experimental.TimeoutExceptionCoroutine.run(Scheduled.kt:110)at kotlinx.coroutines.experimental.EventLoopImpl$DelayedRunnableTask.invoke(EventLoop.kt:199)at kotlinx.coroutines.experimental.EventLoopImpl$DelayedRunnableTask.invoke(EventLoop.kt:195)at kotlinx.coroutines.experimental.EventLoopImpl.processNextEvent(EventLoop.kt:111)at kotlinx.coroutines.experimental.BlockingCoroutine.joinBlocking(Builders.kt:205)at kotlinx.coroutines.experimental.BuildersKt.runBlocking(Builders.kt:150)at kotlinx.coroutines.experimental.BuildersKt.runBlocking$default(Builders.kt:142)at com.easy.kotlin.CancellingCoroutineDemo.testTimeouts(CancellingCoroutineDemo.kt:169)at com.easy.kotlin.CancellingCoroutineDemoKt.main(CancellingCoroutineDemo.kt:193)由 withTimeout 拋出的 TimeoutException 是 CancellationException 的一個子類。這個TimeoutException類型定義如下：
private class TimeoutException(time: Long,unit: TimeUnit,@JvmField val coroutine: Job ) : CancellationException("Timed out waiting for $time $unit")如果您需要在超時時執行一些附加操作, 則可以把邏輯放在 try {…} catch (e: CancellationException) {…} 代碼塊中。例如：
try {ccd.testTimeouts() } catch (e: CancellationException) {println("I am timed out!") }9.8 掛起函數的組合執行
本節我們介紹掛起函數組合的各種方法。
9.8.1 按默認順序執行
假設我們有兩個在別處定義的掛起函數：
suspend fun doJob1(): Int {println("Doing Job1 ...")delay(1000L) // 此處模擬我們的工作代碼println("Job1 Done")return 10 }suspend fun doJob2(): Int {println("Doing Job2 ...")delay(1000L) // 此處模擬我們的工作代碼println("Job2 Done")return 20 }如果需要依次調用它們, 我們只需要使用正常的順序調用, 因為協同中的代碼 (就像在常規代碼中一樣) 是默認的順序執行。下面的示例通過測量執行兩個掛起函數所需的總時間來演示:
fun testSequential() = runBlocking<Unit> {val time = measureTimeMillis {val one = doJob1()val two = doJob2()println("[testSequential] 最終結果： ${one + two}")}println("[testSequential] Completed in $time ms") }執行上面的代碼，我們將得到輸出：
Doing Job1 ... Job1 Done Doing Job2 ... Job2 Done [testSequential] 最終結果： 30 [testSequential] Completed in 6023 ms可以看出，我們的代碼是跟普通的代碼一樣順序執行下去。
9.8.2 使用async異步并發執行
上面的例子中，如果在調用 doJob1 和 doJob2 之間沒有時序上的依賴關系, 并且我們希望通過同時并發地執行這兩個函數來更快地得到答案, 那該怎么辦呢？這個時候，我們就可以使用async來實現異步。代碼示例如下：
fun testAsync() = runBlocking<Unit> {val time = measureTimeMillis {val one = async(CommonPool) { doJob1() }val two = async(CommonPool) { doJob2() }println("最終結果： ${one.await() + two.await()}")}println("Completed in $time ms") }如果跟上面同步的代碼一起執行對比，我們可以看到如下輸出：
Doing Job1 ... Job1 Done Doing Job2 ... Job2 Done [testSequential] 最終結果： 30 [testSequential] Completed in 6023 ms Doing Job1 ... Doing Job2 ... Job1 Done Job2 Done [testAsync] 最終結果： 30 [testAsync] Completed in 3032 ms我們可以看出，使用async函數，我們的兩個Job并發的執行了，并發花的時間要比順序的執行的要快將近兩倍。因為，我們有兩個任務在并發的執行。
從概念上講, async跟launch類似, 它啟動一個協程, 它與其他協程并發地執行。
不同之處在于, launch返回一個任務Job對象, 不帶任何結果值；而async返回一個延遲任務對象Deferred，一種輕量級的非阻塞性future, 它表示后面會提供結果。
在上面的示例代碼中，我們使用Deferred調用 await() 函數來獲得其最終結果。另外，延遲任務Deferred也是Job類型, 它繼承自Job，所以它也有isActive、isCompleted屬性，也有join()、cancel()函數，因此我們也可以在需要時取消它。Deferred接口定義如下：
public interface Deferred<out T> : Job {val isCompletedExceptionally: Booleanval isCancelled: Booleanpublic suspend fun await(): Tpublic fun <R> registerSelectAwait(select: SelectInstance<R>, block: suspend (T) -> R)public fun getCompleted(): T@Deprecated(message = "Use `isActive`", replaceWith = ReplaceWith("isActive"))public val isComputing: Boolean get() = isActive }其中，常用的屬性和函數說明如下：
| isCompletedExceptionally | 當協程在計算過程中有異常failed 或被取消，返回true。 這也意味著isActive等于 false ，同時 isCompleted等于 true |
| isCancelled | 如果當前延遲任務被取消，返回true |
| suspend fun await() | 等待此延遲任務完成，而不阻塞線程；如果延遲任務完成, 則返回結果值或引發相應的異常。 |
延遲任務對象Deferred的狀態與對應的屬性值如下表所示：
| New (可選初始狀態) | false | false | false | false |
| Active (默認初始狀態) | true | false | false | false |
| Resolved (最終狀態) | false | true | false | false |
| Failed (最終狀態) | false | true | true | false |
| Cancelled (最終狀態) | false | true | true | true |
9.9 協程上下文與調度器
到這里，我們已經看到了下面這些啟動協程的方式：
launch(CommonPool) {...} async(CommonPool) {...} run(NonCancellable) {...}這里的CommonPool 和 NonCancellable 是協程上下文（coroutine contexts）。本小節我們簡單介紹一下自定義協程上下文。
9.9.1 調度和線程
協程上下文包括一個協程調度程序, 它可以指定由哪個線程來執行協程。調度器可以將協程的執行調度到一個線程池，限制在特定的線程中；也可以不作任何限制，讓它無約束地運行。請看下面的示例:
fun testDispatchersAndThreads() = runBlocking {val jobs = arrayListOf<Job>()jobs += launch(Unconfined) {// 未作限制 -- 將會在 main thread 中執行println("Unconfined: I'm working in thread ${Thread.currentThread()}")}jobs += launch(context) {// 父協程的上下文 ： runBlocking coroutineprintln("context: I'm working in thread ${Thread.currentThread()}")}jobs += launch(CommonPool) {// 調度指派給 ForkJoinPool.commonPoolprintln("CommonPool: I'm working in thread ${Thread.currentThread()}")}jobs += launch(newSingleThreadContext("MyOwnThread")) {// 將會在這個協程自己的新線程中執行println("newSingleThreadContext: I'm working in thread ${Thread.currentThread()}")}jobs.forEach { it.join() } }運行上面的代碼，我們將得到以下輸出 (可能按不同的順序):
Unconfined: I'm working in thread Thread[main,5,main] CommonPool: I'm working in thread Thread[ForkJoinPool.commonPool-worker-1,5,main] newSingleThreadContext: I'm working in thread Thread[MyOwnThread,5,main] context: I'm working in thread Thread[main,5,main]從上面的結果，我們可以看出：
使用無限制的Unconfined上下文的協程運行在主線程中；
繼承了 runBlocking {…} 的context的協程繼續在主線程中執行；
而CommonPool在ForkJoinPool.commonPool中；
我們使用newSingleThreadContext函數新建的協程上下文，該協程運行在自己的新線程Thread[MyOwnThread,5,main]中。
另外，我們還可以在使用 runBlocking的時候顯式指定上下文, 同時使用 run 函數來更改協程的上下文：
fun log(msg: String) = println("${Thread.currentThread()} $msg")fun testRunBlockingWithSpecifiedContext() = runBlocking {log("$context")log("${context[Job]}")log("開始")val ctx1 = newSingleThreadContext("線程A")val ctx2 = newSingleThreadContext("線程B")runBlocking(ctx1) {log("Started in Context1")run(ctx2) {log("Working in Context2")}log("Back to Context1")}log("結束") }運行輸出：
Thread[main,5,main] [BlockingCoroutine{Active}@b1bc7ed, EventLoopImpl@7cd84586] Thread[main,5,main] BlockingCoroutine{Active}@b1bc7ed Thread[main,5,main] 開始 Thread[線程A,5,main] Started in Context1 Thread[線程B,5,main] Working in Context2 Thread[線程A,5,main] Back to Context1 Thread[main,5,main] 結束9.9.2 父子協程
當我們使用協程A的上下文啟動另一個協程B時, B將成為A的子協程。當父協程A任務被取消時, B以及它的所有子協程都會被遞歸地取消。代碼示例如下：
fun testChildrenCoroutine()= runBlocking<Unit> {val request = launch(CommonPool) {log("ContextA1: ${context}")val job1 = launch(CommonPool) {println("job1: 獨立的協程上下文!")delay(1000)println("job1: 不會受到request.cancel()的影響")}// 繼承父上下文：request的contextval job2 = launch(context) {log("ContextA2: ${context}")println("job2: 是request coroutine的子協程")delay(1000)println("job2: 當request.cancel()，job2也會被取消")}job1.join()job2.join()}delay(500)request.cancel()delay(1000)println("main: Who has survived request cancellation?") }運行輸出：
Thread[ForkJoinPool.commonPool-worker-1,5,main] ContextA1: [StandaloneCoroutine{Active}@5b646af2, CommonPool] job1: 獨立的協程上下文! Thread[ForkJoinPool.commonPool-worker-3,5,main] ContextA2: [StandaloneCoroutine{Active}@75152aa4, CommonPool] job2: 是request coroutine的子協程 job1: 不會受到request.cancel()的影響 main: Who has survived request cancellation?9.10 通道
延遲對象提供了一種在協程之間傳輸單個值的方法。而通道（Channel）提供了一種傳輸數據流的方法。通道是使用 SendChannel 和使用 ReceiveChannel 之間的非阻塞通信。
9.10.1 通道 vs 阻塞隊列
通道的概念類似于 阻塞隊列（BlockingQueue）。在Java的Concurrent包中，BlockingQueue很好的解決了多線程中如何高效安全“傳輸”數據的問題。它有兩個常用的方法如下：
E take(): 取走BlockingQueue里排在首位的對象,若BlockingQueue為空, 阻塞進入等待狀態直到BlockingQueue有新的數據被加入;
put(E e): 把對象 e 加到BlockingQueue里, 如果BlockQueue沒有空間,則調用此方法的線程被阻塞，直到BlockingQueue里面有空間再繼續。
通道跟阻塞隊列一個關鍵的區別是：通道有掛起的操作, 而不是阻塞的, 同時它可以關閉。
代碼示例：
package com.easy.kotlinimport kotlinx.coroutines.experimental.CommonPool import kotlinx.coroutines.experimental.channels.Channel import kotlinx.coroutines.experimental.launch import kotlinx.coroutines.experimental.runBlockingclass ChannelsDemo {fun testChannel() = runBlocking<Unit> {val channel = Channel<Int>()launch(CommonPool) {for (x in 1..10) channel.send(x * x)}println("channel = ${channel}")// here we print five received integers:repeat(10) { println(channel.receive()) }println("Done!")} }fun main(args: Array<String>) {val cd = ChannelsDemo()cd.testChannel() }運行輸出：
channel = kotlinx.coroutines.experimental.channels.RendezvousChannel@2e817b38 1 4 9 16 25 36 49 64 81 100 Done!我們可以看出使用Channel<Int>()背后調用的是會合通道RendezvousChannel()，會合通道中沒有任何緩沖區。send函數被掛起直到另外一個協程調用receive函數， 然后receive函數掛起直到另外一個協程調用send函數。它是一個完全無鎖的實現。
9.10.2 關閉通道和迭代遍歷元素
與隊列不同, 通道可以關閉, 以指示沒有更多的元素。在接收端, 可以使用 for 循環從通道接收元素。代碼示例：
fun testClosingAndIterationChannels() = runBlocking {val channel = Channel<Int>()launch(CommonPool) {for (x in 1..5) channel.send(x * x)channel.close() // 我們結束 sending}// 打印通道中的值，直到通道關閉for (x in channel) println(x)println("Done!") }其中， close函數在這個通道上發送一個特殊的 “關閉令牌”。這是一個冪等運算：對此函數的重復調用不起作用, 并返回 “false”。此函數執行后，isClosedForSend返回 “true”。但是, ReceiveChannel的isClosedForReceive在所有之前發送的元素收到之后才返回 “true”。
我們把上面的代碼加入打印日志：
fun testClosingAndIterationChannels() = runBlocking {val channel = Channel<Int>()launch(CommonPool) {for (x in 1..5) {channel.send(x * x)}println("Before Close => isClosedForSend = ${channel.isClosedForSend}")channel.close() // 我們結束 sendingprintln("After Close => isClosedForSend = ${channel.isClosedForSend}")}// 打印通道中的值，直到通道關閉for (x in channel) {println("${x} => isClosedForReceive = ${channel.isClosedForReceive}")}println("Done! => isClosedForReceive = ${channel.isClosedForReceive}") }運行輸出：
1 => isClosedForReceive = false 4 => isClosedForReceive = false 9 => isClosedForReceive = false 16 => isClosedForReceive = false 25 => isClosedForReceive = false Before Close => isClosedForSend = false After Close => isClosedForSend = true Done! => isClosedForReceive = true9.10.3 生產者-消費者模式
使用協程生成元素序列的模式非常常見。這是在并發代碼中經常有的生產者-消費者模式。代碼示例：
fun produceSquares() = produce<Int>(CommonPool) {for (x in 1..7) send(x * x) }fun consumeSquares() = runBlocking{val squares = produceSquares()squares.consumeEach { println(it) }println("Done!") }這里的produce函數定義如下：
public fun <E> produce(context: CoroutineContext,capacity: Int = 0,block: suspend ProducerScope<E>.() -> Unit ): ProducerJob<E> {val channel = Channel<E>(capacity)return ProducerCoroutine(newCoroutineContext(context), channel).apply {initParentJob(context[Job])block.startCoroutine(this, this)} }其中，參數說明如下：
| context | 協程上下文 |
| capacity | 通道緩存容量大小 (默認沒有緩存) |
| block | 協程代碼塊 |
produce函數會啟動一個新的協程, 協程中發送數據到通道來生成數據流，并以 ProducerJob 對象返回對協程的引用。ProducerJob繼承了Job, ReceiveChannel類型。
9.11 管道
9.11.1 生產無限序列
管道（Pipeline）是一種模式, 我們可以用一個協程生產無限序列:
fun produceNumbers() = produce<Long>(CommonPool) {var x = 1Lwhile (true) send(x++) // infinite stream of integers starting from 1 }我們的消費序列的函數如下：
fun square(numbers: ReceiveChannel<Int>) = produce<Int>(CommonPool) {for (x in numbers) send(x * x) }主代碼啟動并連接整個管線:
fun testPipeline() = runBlocking {val numbers = produceNumbers() // produces integers from 1 and onval squares = consumeNumbers(numbers) // squares integers//for (i in 1..6) println(squares.receive())while (true) {println(squares.receive())}println("Done!")squares.cancel()numbers.cancel() }運行上面的代碼，我們將會發現控制臺在打印一個無限序列，完全沒有停止的意思。
9.11.2 管道與無窮質數序列
我們使用協程管道來生成一個無窮質數序列。
我們從無窮大的自然數序列開始：
fun numbersProducer(context: CoroutineContext, start: Int) = produce<Int>(context) {var n = startwhile (true) send(n++) // infinite stream of integers from start }這次我們引入一個顯式上下文參數context, 以便調用方可以控制我們的協程運行的位置。
下面的管道將篩選傳入的數字流, 過濾掉可以被當前質數整除的所有數字：
fun filterPrimes(context: CoroutineContext, numbers: ReceiveChannel<Int>, prime: Int) = produce<Int>(context) {for (x in numbers) if (x % prime != 0) send(x) }現在我們通過從2開始, 從當前通道中取一個質數, 并為找到的每個質數啟動新的管道階段, 從而構建出我們的管道:
numbersFrom(2) -> filterPrimes(2) -> filterPrimes(3) -> filterPrimes(5) -> filterPrimes(7) ...測試無窮質數序列：
fun producePrimesSequences() = runBlocking {var producerJob = numbersProducer(context, 2)while (true) {val prime = producerJob.receive()print("${prime} \t")producerJob = filterPrimes(context, producerJob, prime)} }運行上面的代碼，我們將會看到控制臺一直在無限打印出質數序列：
9.11.3 通道緩沖區
我們可以給通道設置一個緩沖區：
fun main(args: Array<String>) = runBlocking<Unit> {val channel = Channel<Int>(4) // 創建一個緩沖區容量為4的通道launch(context) {repeat(10) {println("Sending $it")channel.send(it) // 當緩沖區已滿的時候， send將會掛起}}delay(1000) }輸出：
Sending 0 Sending 1 Sending 2 Sending 3 Sending 49.12 構建無窮惰性序列
我們可以使用 buildSequence 序列生成器 ，構建一個無窮惰性序列。
val fibonacci = buildSequence {yield(1L)var current = 1Lvar next = 1Lwhile (true) {yield(next)val tmp = current + nextcurrent = nextnext = tmp} }我們通過buildSequence創建一個協程，生成一個惰性的無窮斐波那契數列。該協程通過調用 yield() 函數來產生連續的斐波納契數。
我們可以從該序列中取出任何有限的數字列表，例如
println(fibonacci.take(16).toList())的結果是：
[1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, 610, 987]9.13 協程與線程比較
直接先說區別，協程是編譯器級的，而線程是操作系統級的。
協程通常是由編譯器來實現的機制。線程看起來也在語言層次，但是內在原理卻是操作系統先有這個東西，然后通過一定的API暴露給用戶使用，兩者在這里有不同。
協程就是用戶空間下的線程。用協程來做的東西，用線程或進程通常也是一樣可以做的，但往往多了許多加鎖和通信的操作。
線程是搶占式，而協程是非搶占式的，所以需要用戶自己釋放使用權來切換到其他協程，因此同一時間其實只有一個協程擁有運行權，相當于單線程的能力。
協程并不是取代線程, 而且抽象于線程之上, 線程是被分割的CPU資源, 協程是組織好的代碼流程, 協程需要線程來承載運行, 線程是協程的資源, 但協程不會直接使用線程, 協程直接利用的是執行器(Interceptor), 執行器可以關聯任意線程或線程池, 可以使當前線程, UI線程, 或新建新程.。
線程是協程的資源。協程通過Interceptor來間接使用線程這個資源。
9.14 協程的好處
與多線程、多進程等并發模型不同，協程依靠user-space調度，而線程、進程則是依靠kernel來進行調度。線程、進程間切換都需要從用戶態進入內核態，而協程的切換完全是在用戶態完成，且不像線程進行搶占式調度，協程是非搶占式的調度。
通常多個運行在同一調度器中的協程運行在一個線程內，這也消除掉了多線程同步等帶來的編程復雜性。同一時刻同一調度器中的協程只有一個會處于運行狀態。
我們使用協程，程序只在用戶空間內切換上下文，不再陷入內核來做線程切換，這樣可以避免大量的用戶空間和內核空間之間的數據拷貝，降低了CPU的消耗，從而大大減緩高并發場景時CPU瓶頸的窘境。
另外，使用協程，我們不再需要像異步編程時寫那么一堆callback函數，代碼結構不再支離破碎，整個代碼邏輯上看上去和同步代碼沒什么區別，簡單，易理解，優雅。
我們使用協程，我們可以很簡單地實現一個可以隨時中斷隨時恢復的函數。
一些 API 啟動長時間運行的操作(例如網絡 IO、文件 IO、CPU 或 GPU 密集型任務等)，并要求調用者阻塞直到它們完成。協程提供了一種避免阻塞線程并用更廉價、更可控的操作替代線程阻塞的方法：協程掛起。
協程通過將復雜性放入庫來簡化異步編程。程序的邏輯可以在協程中順序地表達，而底層庫會為我們解決其異步性。該庫可以將用戶代碼的相關部分包裝為回調、訂閱相關事件、在不同線程(甚至不同機器)上調度執行，而代碼則保持如同順序執行一樣簡單。
9.14.1 阻塞 vs 掛起
協程可以被掛起而無需阻塞線程。而線程阻塞的代價通常是昂貴的，尤其在高負載時，阻塞其中一個會導致一些重要的任務被延遲。
另外，協程掛起幾乎是無代價的。不需要上下文切換或者 OS 的任何其他干預。
最重要的是，掛起可以在很大程度上由用戶來控制，我們可以決定掛起時做些，并根據需求優化、記日志、攔截處理等。
9.15 協程的內部機制
9.15.1 基本原理
協程完全通過編譯技術實現(不需要來自 VM 或 OS 端的支持)，掛起機制是通過狀態機來實現，其中的狀態對應于掛起調用。
在掛起時，對應的協程狀態與局部變量等一起被存儲在編譯器生成的類的字段中。在恢復該協程時，恢復局部變量并且狀態機從掛起點接著后面的狀態往后執行。
掛起的協程，是作為Continuation對象來存儲和傳遞，Continuation中持有協程掛起狀態與局部變量。
關于協程工作原理的更多細節可以在這個設計文檔中找到。
9.15.2 標準 API
協程有三個主要組成部分：
- 語言支持(即如上所述的掛起功能)，
- Kotlin 標準庫中的底層核心 API，
- 可以直接在用戶代碼中使用的高級 API。
- 底層 API：kotlin.coroutines
底層 API 相對較小，并且除了創建更高級的庫之外，不應該使用它。 它由兩個主要包組成：
kotlin.coroutines.experimental 帶有主要類型與下述原語：
- createCoroutine()
- startCoroutine()
- suspendCoroutine()
kotlin.coroutines.experimental.intrinsics 帶有甚至更底層的內在函數如 ：
- suspendCoroutineOrReturn()
大多數基于協程的應用程序級API都作為單獨的庫發布：kotlinx.coroutines。這個庫主要包括下面幾大模塊：
- 使用 kotlinx-coroutines-core 的平臺無關異步編程
- 基于 JDK 8 中的 CompletableFuture 的 API：kotlinx-coroutines-jdk8
- 基于 JDK 7 及更高版本 API 的非阻塞 IO(NIO)：kotlinx-coroutines-nio
- 支持 Swing (kotlinx-coroutines-swing) 和 JavaFx (kotlinx-coroutines-javafx)
- 支持 RxJava：kotlinx-coroutines-rx
這些庫既作為使通用任務易用的便利的 API，也作為如何構建基于協程的庫的端到端示例。關于這些 API 用法的更多細節可以參考相關文檔。
本章小結
本章我通過大量實例學習了協程的用法；同時了解了作為輕量級線程的協程是怎樣簡化的我們的多線程并發編程的。我們看到協程通過掛起機制實現非阻塞的特性大大提升了我們并發性能。
最后，我們還簡單介紹了協程的實現的原理以及標準API庫。Kotlin的協程的實現大量地調用了Java中的多線程API。所以在Kotlin中，我們仍然完全可以使用Java中的多線程編程。
下一章我們來一起學習Kotlin與Java代碼之間的互相調用。
本章示例代碼工程：https://github.com/EasyKotlin/chapter9_coroutines
總結
以上是生活随笔為你收集整理的Kotlin极简教程：第9章 轻量级线程：协程的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 高阶函数（二）
- 下一篇： Kotlin极简教程：第10章 Kotl