Kotlin协程笔记

2023年11月22日 276次阅读来源: dengyin2000

很多小伙伴可能会觉得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

    原文作者：dengyin2000
    原文地址: https://www.jianshu.com/p/8dc8abca50e3
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。

算法网

Kotlin协程笔记

Kotlin语言基础笔记

Kotlin流程控制语句笔记

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

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

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

Kotlin类型系统笔记

Kotlin面向对象编程笔记

Kotlin委托（Delegation）笔记

Kotlin泛型型笔记

Kotlin函数式编程笔记

Kotlin与Java互操作笔记

Kotlin协程笔记

1.第一个协程代码

2. runBlocking

3.等待协程完成

4. suspending function 暂停函数

5. 协程是非常轻量级的

6. 协程像守护线程

7. 协程取消

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

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

10. 协程中try catch finally

11. withContext函数

12. Timeout超时

13. 使用async并发调用

14. async延时调用

15. Async-style functions

16. Dispatchers and threads

17. Unconfined vs confined dispatcher

18. 线程切换

19. 通过Context来获取Job

19. 子协程

20. Context联合

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

22. Tricks

23. channel, select, actor