Kotlin语言基础笔记

Kotlin流程控制语句笔记

Kotlin操作符重载与中缀表示法笔记

Kotlin扩展函数和扩展属性笔记

Kotlin空指针安全（null-safety）笔记

Kotlin类型系统笔记

Kotlin面向对象编程笔记

Kotlin委托（Delegation）笔记

Kotlin泛型型笔记

Kotlin函数式编程笔记

Kotlin与Java互操作笔记

Kotlin协程笔记

很多小伙伴可能会觉得Java有了线程、线程池了，我们还要协程（Coroutines）干嘛。这里还是有些区别的。区别有：

• 线程是为了提高CPU的利用率，调度是由操作系统决定的，而协程是为了解决多个任务更好的协作，调度是由我们代码控制的。
• 协程并不是为了取代线程，协程对线程进行抽象，你可以看成协程是一个异步调用的框架，解决了之前线程间协作代码繁琐的问题。

我们先来看一段代码，如下：

data class Product(var id: String, var title: String)
data class Stock(var pid: String, var stock: Int)
data class Pms(var pid: String, var pmsTips: String)

suspend fun getProductsByIds(pids: List<String>): List<Product> {
    delay(1000)
    return listOf(Product("1", "a"), Product("2", "b"))
}

suspend fun getProductStocksByIds(pids: List<String>): List<Stock> {
    delay(2000)
    return listOf(Stock("1", 2), Stock("2", 4))
}

suspend fun getProductPMSByIds(pids: List<String>): List<Pms> {
    delay(3000)
    return listOf(Pms("1", "100减99"), Pms("2", "100减99"))
}

fun combine(products: List<Product>?, productStocks: List<Stock>?, productPMS: List<Pms>?) {
    println(products)
    println(productStocks)
    println(productPMS)
}

fun main(args: Array<String>) = runBlocking<Unit> {
    val pids = listOf<String>("1", "2")
    val products = async {
        withTimeoutOrNull(1500) {
            getProductsByIds(pids)
        }
    }
    val productStocks = async {
        withTimeoutOrNull(2500) {
            getProductStocksByIds(pids)
        }
    }
    val productPMS = async {
        withTimeoutOrNull(2500) {
            getProductPMSByIds(pids)
        }
    }

    val measureTimeMillis = measureTimeMillis {
        combine(products.await(), productStocks.await(), productPMS.await())
    }
    println(measureTimeMillis)
}

这段代码看起来就像是伪代码，不过还是非常容易理解，就是通过一批商品id，分别调用三个接口拿到商品的信息，商品的库存，商品的优惠信息，然后再合并数据，这个场景无论在后端还是前端都会经常遇到，比如APP调用的一个接口，需要从不同的底层系统获取到不同部分的数据，然后聚合好一次性返回给APP。想想如果是用Java来实现的会有多复杂。用Kotlin的协程实现就像是写顺序执行的代码，但实际上你做的是异步调用。

1.第一个协程代码

fun main(args: Array<String>) {
    launch { // launch new coroutine in background and continue
        delay(1000L) // non-blocking delay for 1 second (default time unit is ms)
        println("World!") // print after delay
    }
    println("Hello,") // main thread continues while coroutine is delayed
    Thread.sleep(2000L) // block main thread for 2 seconds to keep JVM alive
}

我们使用lauch来启动一个协程，其中要注意的是delay这个函数，看起来它跟Thread.sleep是一样的作用，但是他们有本质的区别，Thread.sleep会阻塞当前线程（线程就傻傻的在等待），而delay是暂停当前的协程，不会阻塞当前线程，这个线程可以去做其他事情。delay是一个suspending function，它只能运行在协程里面，如果不在协程中运行，会报以下异常。

Error: Kotlin: Suspend functions are only allowed to be called from a coroutine or another suspend function

2. runBlocking

runBlocking函数会阻塞当前线程，一直等到协程运行完。上面的例子可以改成下面的：

fun main(args: Array<String>) = runBlocking<Unit> { // start main coroutine
    launch { // launch new coroutine in background and continue
        delay(1000L)
        println("World!")
    }
    println("Hello,") // main coroutine continues here immediately
    delay(2000L)      // delaying for 2 seconds to keep JVM alive
}

3.等待协程完成

延时一段时间来等待协程完成通常不是很高效，我们可以通过join来实现一旦协程完成就退出main函数。

fun main(args: Array<String>) = runBlocking<Unit> {
    val job = launch { // launch new coroutine and keep a reference to its Job
        delay(1000L)
        println("World!")
    }
    println("Hello,")
    job.join() // wait until child coroutine completes
}

4. suspending function 暂停函数

我们也可以使用suspending function重构下。

fun main(args: Array<String>) = runBlocking<Unit> {
    val job = launch { doWorld() }
    println("Hello,")
    job.join()
}

// this is your first suspending function
suspend fun doWorld() {
    delay(1000L)
    println("World!")
}

注意：delay也是一个suspending function，所以depay只能放在suspending function或者协程代码（lanuch）里面。

5. 协程是非常轻量级的

fun main(args: Array<String>) = runBlocking<Unit> {
    val jobs = List(100_000) { // launch a lot of coroutines and list their jobs
        launch {
            delay(1000L)
            print(".")
        }
    }
    jobs.forEach { it.join() } // wait for all jobs to complete
}

启动了10万个协程，最后代码能够成功的执行完成。同样，大家可以试试换成起10万个线程试试，应该会得出OOM的结果。

6. 协程像守护线程

请看下面这段代码：

fun main(args: Array<String>) = runBlocking<Unit> {
    launch {
        repeat(1000) { i ->
            println("I'm sleeping $i ...")
            delay(500L)
        }
    }
    delay(1300L) // just quit after delay
}

输出如下：

I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...

可以知道，等待1.3秒后，main退出了。不会等待launch的协程运行完。

7. 协程取消

launch返回一个Job对象，它可以被取消：

fun main(args: Array<String>) = runBlocking<Unit> {
    val job = launch {
        repeat(1000) { i ->
            println("I'm sleeping $i ...")
            delay(500L)
        }
    }
    delay(1300L) // delay a bit
    println("main: I'm tired of waiting!")
    job.cancel() // cancels the job
    job.join() // waits for job's completion 
    println("main: Now I can quit.")
}

输出如下：

I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
main: I'm tired of waiting!
main: Now I can quit.

可以看到，一旦调用了job.cancel()，就退出了main函数。Job还有一个cancelAndJoin方法，合并了cancel和join操作。

8. 协程的取消可能需要协作完成

协程的取消可能需要协作完成，所有在kotlinx.coroutines包下面的suspending functions都可以被取消，但是如果一个协程处在计算中，他是不能被取消的，比如这个例子：

fun main(args: Array<String>) = runBlocking<Unit> {
    val startTime = System.currentTimeMillis()
    val job = launch {
        var nextPrintTime = startTime
        var i = 0
        while (i < 5) { // computation loop, just wastes CPU
            // print a message twice a second
            if (System.currentTimeMillis() >= nextPrintTime) {
                println("I'm sleeping ${i++} ...")
                nextPrintTime += 500L
            }
        }
    }
    delay(1300L) // delay a bit
    println("main: I'm tired of waiting!")
    job.cancelAndJoin() // cancels the job and waits for its completion
    println("main: Now I can quit.")
}

你可以看到调用取消后，还在打印。

9. 让处于计算中的协程可取消

有两种方式可以做到：

• 最简单的在while循环最后面调用下yield函数。这样就在每次循环后让协程有了被取消的机会，yield是kotlinx.coroutines包下的suspending functions。
• 检查协程取消的状态，如果发现被取消，则退出循环。
  下面我们以第二种方式演示下：

fun main(args: Array<String>) = runBlocking<Unit> {
    val startTime = System.currentTimeMillis()
    val job = launch {
        var nextPrintTime = startTime
        var i = 0
        while (isActive) { // cancellable computation loop
            // print a message twice a second
            if (System.currentTimeMillis() >= nextPrintTime) {
                println("I'm sleeping ${i++} ...")
                nextPrintTime += 500L
            }
        }
    }
    delay(1300L) // delay a bit
    println("main: I'm tired of waiting!")
    job.cancelAndJoin() // cancels the job and waits for its completion
    println("main: Now I can quit.")

isActive是协程的CoroutineScope的一个属性。

10. 协程中try catch finally

当协程被取消时，catch和finally可以被执行。

fun main(args: Array<String>) = runBlocking<Unit> {
    val job = launch {
        try {
            repeat(1000) { i ->
                println("I'm sleeping $i ...")
                delay(500L)
            }
        }catch (e:Throwable){
            println("I'm running catch")
        } finally {
            println("I'm running finally")
        }
    }
    delay(1300L) // delay a bit
    println("main: I'm tired of waiting!")
    job.cancelAndJoin() // cancels the job and waits for its completion
    println("main: Now I can quit.")
}

输出：

I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
main: I'm tired of waiting!
I'm running catch
I'm running finally
main: Now I can quit.

11. withContext函数

在上个例子中，如果我们在finally块中调用suspending functions的话，会抛出CancellationException，因为协程已经被取消了。不过一般来说没什么太大问题，只要不调用suspending functions。如果你一定要在调用的话，你可以使用withContext(NonCancellable) {...}。如下：

fun main(args: Array<String>) = runBlocking<Unit> {
    val job = launch {
        try {
            repeat(1000) { i ->
                println("I'm sleeping $i ...")
                delay(500L)
            }
        } finally {
            withContext(NonCancellable) {
                println("I'm running finally")
                delay(1000L)
                println("And I've just delayed for 1 sec because I'm non-cancellable")
            }
        }
    }
    delay(1300L) // delay a bit
    println("main: I'm tired of waiting!")
    job.cancelAndJoin() // cancels the job and waits for its completion
    println("main: Now I can quit.")
}

12. Timeout超时

如果要设定协程调用超时时间，我们可以使用withTimeout函数，如下：

fun main(args: Array<String>) = runBlocking<Unit> {
    withTimeout(1300L) {
        repeat(1000) { i ->
            println("I'm sleeping $i ...")
            delay(500L)
        }
    }
}

输出如下：

I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
Exception in thread "main" kotlinx.coroutines.experimental.TimeoutCancellationException: Timed out waiting for 1300 MILLISECONDS

如果超时的时候你不想抛出异常，你可以使用withTimeoutOrNull函数，超时的时候它会返回null。

fun main(args: Array<String>) = runBlocking<Unit> {
    val result = withTimeoutOrNull(1300L) {
        repeat(1000) { i ->
            println("I'm sleeping $i ...")
            delay(500L)
        }
        "Done" // will get cancelled before it produces this result
    }
    println("Result is $result")
}

输出如下：

I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
Result is null

13. 使用async并发调用

async与launch类似，它也是启动一个协程，只不过lauch返回的是Job（没有返回值），而async返回的是Deferred（带返回值），你可以使用.await()来获取Deferred的值。Deferred是Job的子类，所以Deferred也可以被取消。看看下面这段代码：

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // pretend we are doing something useful here
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // pretend we are doing something useful here, too
    return 29
}

fun main(args: Array<String>) = runBlocking<Unit> {
    val time = measureTimeMillis {
        val one = async { doSomethingUsefulOne() }
        val two = async { doSomethingUsefulTwo() }
        println("The answer is ${one.await() + two.await()}")
    }
    println("Completed in $time ms")
}

输出如下：

The answer is 42
Completed in 1016 ms

因为是并行调用，所以时间差不多是1秒。

14. async延时调用

fun main(args: Array<String>) = runBlocking<Unit> {
    val time = measureTimeMillis {
        val one = async(start = CoroutineStart.LAZY) { doSomethingUsefulOne() }
        val two = async(start = CoroutineStart.LAZY) { doSomethingUsefulTwo() }
        println("The answer is ${one.await() + two.await()}")
    }
    println("Completed in $time ms")
}

如果async带上了start = CoroutineStart.LAZY参数，协程不会立即执行，会等到调用await的时候才开始执行。上面代码输出如下：

The answer is 42
Completed in 2017 ms

执行结果看起来变成了顺序执行，那是因为one.await执行完成之后，才会开始调用two.await()执行。所以变成了顺序执行。

15. Async-style functions

// The result type of somethingUsefulOneAsync is Deferred<Int>
fun somethingUsefulOneAsync() = async {
    doSomethingUsefulOne()
}

// The result type of somethingUsefulTwoAsync is Deferred<Int>
fun somethingUsefulTwoAsync() = async {
    doSomethingUsefulTwo()
}

上面两个方法xxxAsync并不是suspending functions，所以他们可以在任何地方调用。

// note, that we don't have `runBlocking` to the right of `main` in this example
fun main(args: Array<String>) {
    val time = measureTimeMillis {
        // we can initiate async actions outside of a coroutine
        val one = somethingUsefulOneAsync()
        val two = somethingUsefulTwoAsync()
        // but waiting for a result must involve either suspending or blocking.
        // here we use `runBlocking { ... }` to block the main thread while waiting for the result
        runBlocking {
            println("The answer is ${one.await() + two.await()}")
        }
    }
    println("Completed in $time ms")
}

16. Dispatchers and threads

launch和async都接收一个可选的CoroutineContext参数可以用来指定CoroutineDispatcher。如下：

fun main(args: Array<String>) = runBlocking<Unit> {
    val jobs = arrayListOf<Job>()
    jobs += launch(Unconfined) { // not confined -- will work with main thread
        println("      'Unconfined': I'm working in thread ${Thread.currentThread().name}")
    }
    jobs += launch(coroutineContext) { // context of the parent, runBlocking coroutine
        println("'coroutineContext': I'm working in thread ${Thread.currentThread().name}")
    }
    jobs += launch(CommonPool) { // will get dispatched to ForkJoinPool.commonPool (or equivalent)
        println("      'CommonPool': I'm working in thread ${Thread.currentThread().name}")
    }
    jobs += launch(newSingleThreadContext("MyOwnThread")) { // will get its own new thread
        println("          'newSTC': I'm working in thread ${Thread.currentThread().name}")
    }
    jobs.forEach { it.join() }
}

输出如下：

      'Unconfined': I'm working in thread main
      'CommonPool': I'm working in thread ForkJoinPool.commonPool-worker-1
          'newSTC': I'm working in thread MyOwnThread
'coroutineContext': I'm working in thread main

默认的dispatcher是DefaultDispatcher当前的实现是CommonPool

17. Unconfined vs confined dispatcher

Unconfined dispatcher会在当前线程开始执行协程，但是仅仅是在第一个暂停点，之后它恢复后的dispatcher取决于那个线程执行suspending function。

coroutineContext 是CoroutineScope的一个属性，它的dispatcher会继承它parent线程的dispatcher。 代码如下：

fun main(args: Array<String>) = runBlocking<Unit> {
    val jobs = arrayListOf<Job>()
    jobs += launch(Unconfined) { // not confined -- will work with main thread
        println("      'Unconfined': I'm working in thread ${Thread.currentThread().name}")
        delay(500)
        println("      'Unconfined': After delay in thread ${Thread.currentThread().name}")
    }
    jobs += launch(coroutineContext) { // context of the parent, runBlocking coroutine
        println("'coroutineContext': I'm working in thread ${Thread.currentThread().name}")
        delay(1000)
        println("'coroutineContext': After delay in thread ${Thread.currentThread().name}")
    }
    jobs.forEach { it.join() }
}

输出如下：

      'Unconfined': I'm working in thread main
'coroutineContext': I'm working in thread main
      'Unconfined': After delay in thread kotlinx.coroutines.DefaultExecutor
'coroutineContext': After delay in thread main

所以，coroutineContext继承了runBlocking的main线程，而unconfined恢复后变成了default executor线程。

18. 线程切换

加上-Dkotlinx.coroutines.debugJVM参数运行下面的代码：

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")

fun main(args: Array<String>) {
    newSingleThreadContext("Ctx1").use { ctx1 ->
        newSingleThreadContext("Ctx2").use { ctx2 ->
            runBlocking(ctx1) {
                log("Started in ctx1")
                withContext(ctx2) {
                    log("Working in ctx2")
                }
                log("Back to ctx1")
            }
        }
    }
}

这里展示了几个用法：一个是使用runBlocking指明一个特殊的Context，另外一个是使用withContext来切换Context，输出如下：

[Ctx1 @coroutine#1] Started in ctx1
[Ctx2 @coroutine#1] Working in ctx2
[Ctx1 @coroutine#1] Back to ctx1

还有就是run来释放线程。

19. 通过Context来获取Job

协程的Job是Context的一个属性，如下：

fun main(args: Array<String>) = runBlocking<Unit> {
    println("My job is ${coroutineContext[Job]}")
}

19. 子协程

在协程中使用coroutineContext来启动另一个协程，新协程的Job变成了父协程的子Job，当父协程取消时，子协程也会被取消。

fun main(args: Array<String>) = runBlocking<Unit> {
    // launch a coroutine to process some kind of incoming request
    val request = launch {
        // it spawns two other jobs, one with its separate context
        val job1 = launch {
            println("job1: I have my own context and execute independently!")
            delay(1000)
            println("job1: I am not affected by cancellation of the request")
        }
        // and the other inherits the parent context
        val job2 = launch(coroutineContext) {
            delay(100)
            println("job2: I am a child of the request coroutine")
            delay(1000)
            println("job2: I will not execute this line if my parent request is cancelled")
        }
        // request completes when both its sub-jobs complete:
        job1.join()
        job2.join()
    }
    delay(500)
    request.cancel() // cancel processing of the request
    delay(1000) // delay a second to see what happens
    println("main: Who has survived request cancellation?")
}

输出结果如下：

job1: I have my own context and execute independently!
job2: I am a child of the request coroutine
job1: I am not affected by cancellation of the request
main: Who has survived request cancellation?

20. Context联合

协程Context可以使用+联合，如下：

fun main(args: Array<String>) = runBlocking<Unit> {
    // start a coroutine to process some kind of incoming request
    val request = launch(coroutineContext) { // use the context of `runBlocking`
        // spawns CPU-intensive child job in CommonPool !!! 
        val job = launch(coroutineContext + CommonPool) {
            println("job: I am a child of the request coroutine, but with a different dispatcher")
            delay(1000)
            println("job: I will not execute this line if my parent request is cancelled")
        }
        job.join() // request completes when its sub-job completes
    }
    delay(500)
    request.cancel() // cancel processing of the request
    delay(1000) // delay a second to see what happens
    println("main: Who has survived request cancellation?")
}

job是request的子协程，但是是在CommonPool的线程中执行操作。所以取消request，job也会取消。

21. 父协程会等待子协程完成

父协程会等待子协程完成，不需要使用join来等待他们完成。

fun main(args: Array<String>) = runBlocking<Unit> {
    // launch a coroutine to process some kind of incoming request
    val request = launch {
        repeat(3) { i -> // launch a few children jobs
            launch(coroutineContext)  {
                delay((i + 1) * 200L) // variable delay 200ms, 400ms, 600ms
                println("Coroutine $i is done")
            }
        }
        println("request: I'm done and I don't explicitly join my children that are still active")
    }
    request.join() // wait for completion of the request, including all its children
    println("Now processing of the request is complete")
}

输出如下：

request: I'm done and I don't explicitly join my children that are still active
Coroutine 0 is done
Coroutine 1 is done
Coroutine 2 is done
Now processing of the request is complete

22. Tricks

假如我们现在在写一个anroid app，在activity中启动了很多协程异步调用接口获取数据，当这个activity被destory后，所有的协程需要被取消，要不然就可能会发生内存泄漏。
我们可以创建一个Job实例，然后使用launch(coroutineContext, parent = job)来明确指定parent job。
这样的话，我们可以调用Job.cancel来取消所有的子协程，而Job.join可以等待所有的子协程完成。如下：

fun main(args: Array<String>) = runBlocking<Unit> {
    val job = Job() // create a job object to manage our lifecycle
    // now launch ten coroutines for a demo, each working for a different time
    val coroutines = List(10) { i ->
        // they are all children of our job object
        launch(coroutineContext, parent = job) { // we use the context of main runBlocking thread, but with our parent job
            delay((i + 1) * 200L) // variable delay 200ms, 400ms, ... etc
            println("Coroutine $i is done")
        }
    }
    println("Launched ${coroutines.size} coroutines")
    delay(500L) // delay for half a second
    println("Cancelling the job!")
    job.cancelAndJoin() // cancel all our coroutines and wait for all of them to complete
}

输出如下：

Launched 10 coroutines
Coroutine 0 is done
Coroutine 1 is done
Cancelling the job!

23. channel, select, actor

请看：https://github.com/Kotlin/kotlinx.coroutines/blob/master/coroutines-guide.md