Kotlin协程（入门向）

2020-08-06 本文已影响0人 littlefogcat

最近在学习kotlin的协程，分享一下学习经验！

〇、什么是协程？

官方解释：

协程是轻量级的线程。

个人理解：
~~协程是Kotlin中的线程池。~~

一、如何使用

1. 添加依赖

build.gradle中加入

    implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:1.2.1"
    implementation "org.jetbrains.kotlinx:kotlinx-coroutines-android:1.2.1"

2. 使用

本章只介绍协程的基本用法，并和传统的回调方式、线程池、Rxjava等进行简单的对比。

2.1 情景1：基本用法

当前线程是主线程。我们需要从url中获取到数据，并加载到textView中。

使用回调

    HttpUtil.get(url, object : HttpUtil.Callback {
        override fun onResponse(s: String) {
            runOnUiThread {
                textView.text = s
            }
        }
    })

使用协程

    GlobalScope.launch { // 开启协程
        val response = HttpUtil.get(url)  // 同步请求
        withContext(Dispatchers.Main) { // 切换到主线程，效果等同于runOnUiThread
            textView.text = response
        }
    }

在这里，使用GlobalScope.launch函数开启协程。在其中，使用withContext(Dispatchers.Main)将线程切换到主线程，进行UI操作。这样就完成了一个最简单的协程实例。
这个例子中，协程并没有比回调方式简洁很多。但是接下来的例子中，呈现了一种被称为“回调地狱”的场景。

2.2 情景2：回调地狱

简单列举一个情形：

我们需要调用url1以获取到url2所需要的参数，再调用url2以获取到url3所需要的参数，最后调用url3获取到所需要的数据。
也就是说，必须要等到url1的请求结果出来再请求url2，再等到url2的请求结果出来再请求url3，最后请求url3的返回结果才是真实所需要的数据。

传统方式（使用回调）

    HttpUtil.get(url1, object : HttpUtil.Callback {
        override fun onResponse(response: String) {
            HttpUtil.get(url2, mapOf(Pair("param", response)), object : HttpUtil.Callback {
                override fun onResponse(response: String) {
                    HttpUtil.get(url3, mapOf(Pair("param", response)), object : HttpUtil.Callback {
                        override fun onResponse(response: String) {
                            runOnUiThread {
                                textView.text = response
                            }
                        }
                    })
                }
            })
        }
    })

太可怕了！当然，我们可以使用高阶函数来代替匿名类来优化一下这段代码：

    HttpUtil.get(url1) {
        HttpUtil.get(url2, makeParam(it)) {
            HttpUtil.get(url3, makeParam(it)) { response ->
                runOnUiThread() {
                    textView.text = response
                }
            }
        }
    }

虽然简洁了很多，但是这么多花括号，看着还是挺不爽的！

使用协程

    GlobalScope.launch {
        val response1 = HttpUtil.get(url1)
        val response2 = HttpUtil.get(url2, mapOf(Pair("param", response1)))
        val response3 = HttpUtil.get(url3, mapOf(Pair("param", response2)))
        withContext(Dispatchers.Main) {
            textView.text = response3
        }
    }

可以看到，在这种情形下使用协程，可以使代码变得整洁许多，并且逻辑变得非常清晰。
既然回调方式可以通过高阶函数优化，协程同样有优化的方式。我们可以改造一下HttpUtil.get方法，其在IO线程中执行：

    // HttpUtil.get
    suspend fun get(url: String): String {
        return withContext(Dispatchers.IO) {
            ...
        }
    }

这里使用了suspend关键字，表示这个函数会将协程挂起；换句话说，这个函数是耗时函数。
这样一来，情景2使用协程的代码就可以这么写了：

    GlobalScope.launch(Dispatchers.Main) {
        val response1 = HttpUtil.get(url1)
        val response2 = HttpUtil.get(url2, makeParam(response1))
        val response3 = HttpUtil.get(url3, makeParam(response2))
        textView.text = response3
    }

省去了切换线程的代码后，是不是更简洁了？对比一下回调方式，不得不说协程真香！

2.3 情景3：合并请求结果

我们需要分别请求url1和url2获取到需要的数据，并以此二者返回值为参数调用url3获得最终的数据。

使用回调
这种情景下，使用回调方式，需要用到CountDonwLatch，它几乎是为了这种情况量身定制的：

    val countDownLatch = CountDownLatch(2)
    var param1 = ""
    var param2 = ""
    HttpUtil.get(url1) {
        param1 = it
        countDownLatch.countDown()
    }
    HttpUtil.get(url2) {
        param2 = it
        countDownLatch.countDown()
    }
    thread {
        countDownLatch.await()
        HttpUtil.get(url3, makeParam(param1, param2)) { response ->
            runOnUiThread() {
                textView.text = response
            }
        }
    }

同时，我们可以通过一些外力来实现这个功能，比如RxJava或者线程池。
使用RxJava

    Observable.zip<String, String, String>(
        Observable.create<String> { it.onNext(HttpUtil.get(url1)) }
            .subscribeOn(Schedulers.io()),
        Observable.create<String> { it.onNext(HttpUtil.get(url2)) }
            .subscribeOn(Schedulers.io()),
        BiFunction { param1, param2 ->
            HttpUtil.get(url3, makeParam(param1, param2))
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe { response ->
            textView.text = response
        }

使用线程池

    val executor = Executors.newCachedThreadPool()
    // 因为future.get()会阻塞线程，所以不能在主线程中执行
    executor.execute {
        val param1Future = executor.submit(Callable { HttpUtil.get(url1) })
        val param2Future = executor.submit(Callable { HttpUtil.get(url2) })
        val params = makeParam(param1Future.get(), param2Future.get())
        val response = executor.submit(Callable { HttpUtil.get(url3, params) }).get()
        runOnUiThread {
            textView.text = response
        }
    }

而协程可以更简洁的处理这种情形：
使用协程

    GlobalScope.launch(Dispatchers.IO) {
        val param1 = async { HttpUtil.get(url1) }
        val param2 = async { HttpUtil.get(url2) }
        val response3 = HttpUtil.get(url3, makeParam(param1.await(), param2.await()))
        withContext(Dispatchers.Main) {
            textView.text = response3
        }
    }

可以看到，使用协程的代码非常简洁清晰。
这里使用了async函数，其中的代码块会异步执行，不阻塞当前线程，并返回一个Deferred对象，相当于线程池的Future；而之后再调用await函数，获取其执行结果，这一步是阻塞的，相当于线程池中的Future.get()。
如果不使用async的话，代码会顺序阻塞执行，而不是并发执行了。

3. 协程与线程池

在情景2.3中，可以看到，协程与Java线程池的使用方式非常的相似。
其实，协程的底层就是使用线程池实现的，不过并不是Java中的线程池，而是Kotlin自己实现的线程池。

性能对比

先说结论：协程的多线程执行速度并不会比线程池更快。

协程与线程池

如图所示，在任务数量为10万时，使用Executors.newCachedThreadPool()只用了20秒就执行完了所有任务，当然代价则是CPU稳稳的100%，电脑几近卡死。而使用协程，以Dispatcher.IO作为调度器，执行任务的总时间达到了惊人的3万秒；不过虽然慢是慢了点，线程数最高只有104，占用资源少。
也就是说，比起线程池来讲，协程更轻量级一点，占用更少的资源，而代价是更低的效率。对于Android开发来说，其实很少遇到超高并发的场景，
当然，我们可以使用线程池作为自定义调度器，不过这样做不是画蛇添足么？

不过按照源码注释中的说法，Dispatchers.IO默认最大线程数量为64或者cpu核心数。至于为什么到了104，我也不知道，或许是。
我们可以通过以下代码修改这个最大线程数，比如修改成1000：

    System.setProperty(IO_PARALLELISM_PROPERTY_NAME, "1000")

将最大线程数修改为1000之后，又执行了一下10万任务挑战：

协程与线程池

可以看到，执行时间的确缩短了很多。

二、一些详细说明

1. 一些概念

CoroutineContext
CoroutineContext直译过来是协程上下文，表示一个协程的上下文环境，和Android中的Context类似。它包含了一系列的元素集合，其中最主要的是Job。

Job
Job是一个接口，继承自CoroutineContext.Element。一个Job代表了一项后台任务，每一个协程对应了一个Job。通过launch函数与async函数创建协程，都会返回一个Job对象，通过这个Job对象，可以管理这个协程。Job接口定义了包括但不限于start（启动相关联的协程）、cancel（取消任务）、join（挂起所在的协程直到当前任务完成）等函数。
简单的来说，我们创建协程就是为了完成某项任务，而Job就对应了这项任务。

示例：

fun main() = runBlocking {
    val job1 = launch {
        delay(1000)
        println("job1")
    }
    val job2 = launch {
        delay(500)
        println("job2")
    }
    println("flag 1")
    job2.cancel()
    job1.join()
    println("flag 2")
}

输出：

flag 1
job1
flag 2

因为job2.cancel()取消了job2，所以没有输出job2；而job1.join()挂起了当前协程，所以直到job1输出之后，才输出flag2。

和Java中的线程池作类比的话，这里的Job类似于Java线程池中的Future。launch函数类似于Java线程池中的submit(runnable)，而async则类似于submit(callable)。

CoroutineScope
CoroutineScope直译为协程作用域。所有的协程创建函数，比如我们平时使用的launch、async，都是CoroutineScope的扩展函数。
这么说还是让人很困惑，所以这玩意儿到底有啥用？我查看了许多人的博客，没有一个人说清楚这点的。

最后，还是看了官方的文档才理解。
每个协程都对应了一个CoroutineScope（作用域）。CoroutineScope包含了协程的上下文、Job、子协程等。通过扩展函数launch、async、cancel等，实现了开启子协程、取消所有子任务等功能。在这个作用域下新开启的协程，则是当前协程的子协程。CoroutineScope可以管理协程的生命周期。~~不如把CoroutineScope译作协程管家好了。~~
对于Android开发来说，在Activity中使用协程，会遇到这种情况：当Activity需要销毁的时候，如果协程继续执行，那么就会造成内存泄漏。
有了CoroutineScope，这个问题就很好解决了。首先在Activity中创建一个最高级的CoroutineScope，当需要使用协程的时候，都通过这个作用域来创建协程。这样，所有协程都是这个作用域下的子协程。当销毁Activity时，只需要调用其cancel()函数，就可以取消所有正在执行的任务了。

    // inside an Activity
    val mainScope = MainScope()

    fun someNetwork() {
        mainScope.launch {
            //...
        }
    }

    override fun onDestroy() {
        super.onDestroy()
        mainScope.cancel()
    }

调度器

协程上下文中包含了一个调度器，它限制了协程在哪些线程中执行。
在第一章的例子中，有使用到withContext(Dispatchers.IO)，这其中的Dispatchers.IO就是调度器。

Kotlin内置了四种调度器：

Dispatchers.Default
默认的调度器，基于JVM上的共享线程池，最大线程数为CPU核心数。
Dispatchers.IO
专为IO操作设计的调度器，默认最大线程数为64与CPU核心数的较大值。
Dispatchers.Main
UI主线程调度器。
Dispatchers.Unconfined
无限制调度器。在第一个挂起点之前，在调用它的线程中执行；之后由该挂起函数决定。

2. 不建议使用GlobalScope

GlobalScope是一个特殊的全局CoroutineScope，它不与任何Job绑定。GlobalScope只应该使用在生命周期与整个应用程序相同、且不被取消的协程中。

在第一章中的例子中，我使用了GlobalScope.launch来启动一个协程，这是不被建议的。由于GlobalScope不与任何Job绑定，所以通过它创建的协程无法取消；当在Activity中使用，这很可能会导致内存泄漏。
取而代之的，应该使用非全局的CoroutineScope。

class CoroutineScopeActivity : AppCompatActivity() {
    val mainScope = MainScope() // 非全局的CoroutineScope

    fun someNetwork() {
        mainScope.launch {
            //...
        }
    }

    override fun onDestroy() {
        super.onDestroy()
        mainScope.cancel() // 当Activity销毁时，取消所有任务
    }
}