Kotlin协程的使用

2023-10-16 11:02:52

前言

本篇是在Android官网对Kotlin协程的学习记录。记录Kotlin Coroutines在Android上的特点、应用等

协程概述

一、协程是什么？

协程是一种并发的设计模式，可以使用它来简化异步执行的代码，它可以帮助管理一些耗时的任务，以防耗时任务阻塞主线程。协程可以用同步的方式写出异步代码，代替了传统的回调方式，让代码更具有可读性。

二、协程的特点？

  • 轻量（Lightweight）：其实这里的轻量是相对线程阻塞而言的，协程支持挂起，挂起的时候并不会阻塞当前线程，也就是"非阻塞式挂起"，在协程挂起的时候线程可以做其它的事情，而线程的阻塞期间是无法做其他事情的。所以协程的"非阻塞式挂起"可以节省系统的资源。

  • 内存泄漏更少（Fewer memory leaks）：用户关闭页面的时候，后台线程可能仍然有还在运行的任务，如果使用传统的线程进行后台请求，可能没有很好的办法让线程及时地停止运行，使用协程的话，可以通过Job::cancel让协程及时地停止运行，并且可以通过协程作用域CoroutineScope对协程进行统一管理，例如对通过CoroutineScope启动的协程统一进行cancel，这种就称作结构化并发，它让我们的程序有更少的协程泄漏，协程泄漏可以看做是一种内存泄露。

  • 内置取消支持（Built-in cancellation support）：Cancellation会自动在运行中的整个协程层次结构内传播。

  • Jetpack和第三方框架支持：一些比如Room、ViewModel等Jetpack组件，第三方框架Retrofit等有对Kt协程提供支持。

关于协程作用域：协程必须运行在CoroutineScope里（协程作用域），一个 CoroutineScope 管理一个或多个相关的协程。例如viewmodel-ktx包下面有 viewModelScope，viewModelScope管理通过它启动的协程，如果viewModel被销毁，那么viewModelScope会自动被取消，通过viewModelScope启动的正在运行的协程也会被取消。

挂起与恢复

协程有suspend和resume两项概念：

  • suspend（挂起）：暂停执行当前协程，并保存所有局部变量。
  • resume（恢复）：用于让已挂起的协程从挂起处继续执行。

协程中有一个suspend关键字，它和刚刚提到的suspend概念要区分一下，刚刚提到的suspend(挂起)是一个概念，而suspend关键字可以修饰一个函数，但是仅这个关键字没有让协程挂起的作用，一般suspend关键字是提醒调用者该函数需要直接或间接地在协程下面运行，起到一个标记与提醒的作用。

suspend关键字的标记与提醒有什么作用？在以前，开发者很难判断一个方法是否是耗时的，如果错误地在主线程调用一个耗时方法，那么会造成主线程卡顿，有了suspend关键字，耗时函数的创建者可以将耗时方法使用suspend关键字修饰，并且在方法内部将耗时代码使用withContext{Dispatchers.IO}等方式放到IO线程等运行，开发者只需要直接或间接地在协程下面调用它即可，这样就可以避免耗时任务在主线程中运行从而造成主线程卡顿了。

下面通过官方的一个例子，对协程的suspend和resume两个概念进行说明：

suspend fun fetchDocs() {                             // Dispatchers.Main
    val result = get("https://developer.android.com") // Dispatchers.IO for `get`
    show(result)                                      // Dispatchers.Main
}

suspend fun get(url: String) = withContext(Dispatchers.IO) { /* ... */ }

我们假设在协程中调用fetchDocs方法，该协程提供了一个主线程环境（如启动协程时通过Dispatchers.Main指定），另外，get方法执行耗时任务，它使用挂起函数withContext{Dispatchers.IO}将耗时任务放到了IO线程中执行。

在fetchDocs方法里，当执行到get方法开始进行网络请求的时候，它会挂起(suspend)所在的协程，当网络请求完成时，get会恢复(resume)已挂起的协程，而不是使用回调通知主线程。

Kotlin使用栈帧(stack frame)管理正在运行的函数以及它的局部变量，当挂起一个协程的时候，系统会复制并保存当前的栈帧以供稍后使用。协程恢复时，会将栈帧从其保存位置复制回来，然后函数再次开始运行。

调度器

Kotlin协程必须运行在dispatcher里面，协程可以将自己suspend，dispatcher负责resume它们。

有下面三种Dispatcher：

  • Dispatchers.Main：在主线程运行协程。
  • Dispatchers.IO：该dispatcher适合执行磁盘或网络I/O，并且经过优化。
  • Dispatchers.Default：该dispatcher适合执行占用大量 CPU 资源的工作(对列表排序和解析JSON)，并且经过优化。

启动协程

有以下两种方式启动协程：

  • launch：启动新协程，launch的返回值为Job，协程的执行结果不会返回给调用方。
  • async：启动新协程，async的返回值为Deferred，Deferred继承至Job，可通过调用Deferred::await获取协程的执行结果，其中await是挂起函数。

在一个常规函数启动协程，通常使用的是launch，因为常规函数无法调用Deferred::await，在一个协程或者挂起函数内部开启协程可以使用async。

launch与async的区别：

  1. launch启动的协程没有返回结果；async启动的协程有返回结果。
  2. launch启动的协程有异常会立即抛出；async启动的协程的异常不会立即抛出，会等到调用Deferred::await的时候才将异常抛出。
  3. async适合于一些并发任务的执行，例如有这样的业务：做两个网络请求，等两个请求都完成后，一起显示请求结果。使用async可以这样实现
interface IUser {
    @GET("/users/{nickname}")
    suspend fun getUser(@Path("nickname") nickname: String): User

    @GET("/users/{nickname}")
    fun getUserRx(@Path("nickname") nickname: String): Observable<User>
}

val iUser = ServiceCreator.create(IUser::class.java)
GlobalScope.launch(Dispatchers.Main) {
    val one = async {
        Log.d(TAG, "one: ${threadName()}")
        iUser.getUser("giagor")
    }
    val two = async {
        Log.d(TAG, "two: ${threadName()}")
        iUser.getUser("google")
    }
    Log.d(TAG, "giagor:${one.await()} , google:${two.await()} ")
}

协程概念

CoroutineScope

CoroutineScope会跟踪它使用launch或async创建的所有协程，可以调用scope.cancel()取消该作用域下所有正在运行的协程。在ktx中，为我们提供了一些已经定义好的CoroutineScope，如ViewModel的viewModelScope，Lifecycle的lifecycleScope，具体可以查看Android KTX | Android Developers。

viewModelScope会在ViewModel的onCleared()方法中被取消

可以自己创捷CoroutineScope，如下：

class MainActivity : AppCompatActivity() {
    val scope = CoroutineScope(Job() + Dispatchers.Main)
    
    override fun onCreate(savedInstanceState: Bundle?) {
    	...
        scope.launch {
            Log.d(TAG, "onCreate: ${threadName()}") // main
            fetchDoc1()
        }
        
        scope.launch { 
            ...
        }
    }
    
    suspend fun fetchDoc1() = withContext(Dispatchers.IO) {...}
    
    override fun onDestroy() {
        scope.cancel()
        super.onDestroy()
    }
}

创建scope的时候，将Job和Dispatcher联合起来，作为一个CoroutineContext，作为CoroutineScope的构造参数。当scope.cancel的时候，通过scope开启的所有协程都会被自动取消，并且之后无法使用scope来开启协程（不会报错但是协程开启无效）。

也可以通过传入CoroutineScope的Job来取消协程：

    val job = Job()
    val scope = CoroutineScope(job + Dispatchers.Main)

    scope.launch {...}
	...
	job.cancel()

使用Job取消了协程，之后也是无法通过scope来开启协程的。

其实查看源码，可以发现CoroutineScope.cancel方法内部就是通过Job进行cancel的：

public fun CoroutineScope.cancel(cause: CancellationException? = null) {
    val job = coroutineContext[Job] ?: error("Scope cannot be cancelled because it does not have a job: $this")
    job.cancel(cause)
}

关于协程的取消后面还会再进行介绍。

Job

当我们使用launch或者async创建一个协程的时候，都会获取到一个Job实例，这个Job实例唯一地标识这个协程，并且管理这个协程地生命周期。Job有点类似Java中的Thread类。

Java中Thread类的部分方法：

Kotlin协程的使用

它可以对所创建的线程进行管理。

Job类还有部分扩展函数如下：

Kotlin协程的使用

CoroutineContext

CoroutineContext使用下面的几种元素定义了协程的行为：

  • Job：控制协程的生命周期。
  • CoroutineDispatcher：将工作分派到适当的线程。
  • CoroutineName：协程的名称，可用于调试。
  • CoroutineExceptionHandler：处理未捕获的异常。

对于在作用域内创建的新协程，系统会为新协程分配一个新的 Job 实例，而从包含协程的作用域继承其他 CoroutineContext 元素。可以通过向 launch 或 async 函数传递新的 CoroutineContext 替换继承的元素。请注意，将 Job 传递给 launch 或 async 不会产生任何效果，因为系统始终会向新协程分配 Job 的新实例。

例如：

val scope = CoroutineScope(Job() + Dispatchers.Main + CoroutineName("Top Scope"))

scope.launch(Dispatchers.IO) {
    Log.d(TAG, "onCreate: ${coroutineContext[CoroutineName]}")
}

D/abcde: onCreate: CoroutineName(Top Scope)

新创建的协程从外部的scope继承了CoroutineName等元素，但注意，CoroutineDispatcher元素被重写了，在新创建的协程里，CoroutineDispatcher元素被指定为Dispatchers.IO。

避免使用GlobalScope

官方文档中，对于不提倡使用GlobalScope，给出了三个原因：

  • （一）Promotes hard-coding values. If you hardcode GlobalScope, you might be hard-coding Dispatchers as well.
  • （二）Makes testing very hard as your code is executed in an uncontrolled scope, you won't be able to control its execution.
  • （三）You can't have a common CoroutineContext to execute for all coroutines built into the scope itself.

关于第二点和第三点的解释如下：我们自己创建的CoroutineScope可以进行结构化并发的操作，例如我们可以调用CoroutineScope.cancel去取消该作用域下所有正在运行的协程，cancel方法如下：

public fun CoroutineScope.cancel(cause: CancellationException? = null) {
    val job = coroutineContext[Job] ?: error("Scope cannot be cancelled because it does not have a job: $this")
    job.cancel(cause)
}

它内部先获取CoroutineContext的Job，然后第哦啊有Job的cancel方法，实现协程的取消。我们手动创建的CoroutineScope的CoroutineContext中都是有Job的，例如：

val scope = CoroutineScope(Job() + Dispatchers.Main + CoroutineName("Top Scope"))

它的构造方法为：

public fun CoroutineScope(context: CoroutineContext): CoroutineScope =
    ContextScope(if (context[Job] != null) context else context + Job())

构造方法中，若传入的CoroutineContext没有Job，则会创建一个Job添加到CoroutineContext中。但是GlobalScope是全局(单例)的，它的CoroutineContext是一个EmptyCoroutineContext，里面没有Job成员

public object GlobalScope : CoroutineScope {
    /**
     * Returns [EmptyCoroutineContext].
     */
    override val coroutineContext: CoroutineContext
        get() = EmptyCoroutineContext
}

我们在调用GlobalScope.launch时，可以指定本次启动的协程的CoroutineContext。当我们在调用GlobalScope.cancel()的时候，会报下面的错误：

java.lang.IllegalStateException: Scope cannot be cancelled because it does not have a job: kotlinx.coroutines.GlobalScope@11b671b

可以看出，报错的原因就是因为GlobalScope没有Job。

协程的取消

官方文档的原话：

Cancellation in coroutines is cooperative, which means that when a coroutine's Job is cancelled, the coroutine isn't cancelled until it suspends or checks for cancellation. If you do blocking operations in a coroutine, make sure that the coroutine is cancellable.

可以得出：

  1. 协程的取消是协作式
  2. 外部对当前正在运行的协程的取消，协程不会立即取消，当下面两种情况之一发生时，协程才会取消
    • 该协程的配合检查，协同进行取消，这和停止一个线程的执行类似（需要线程的配合检查）。
    • 当协程suspend的时候，协程也会被取消。

主动检查

举个例子：

val scope = CoroutineScope(Job() + Dispatchers.IO + CoroutineName("Top Scope"))

bn1.setOnClickListener {
	scope.launch {
        Thread.sleep(2000)
        Log.d(TAG, "onCreate: $isActive")
        Log.d(TAG, "onCreate: ${threadName()},${coroutineContext[CoroutineName]?.name}")
	}
}

bn2.setOnClickListener {
	scope.cancel()
}

假如我们只点击bn1开启协程，但是不点击bn2去取消协程，那么输出为

D/abcde: onCreate: true
D/abcde: onCreate: DefaultDispatcher-worker-1,Top Scope

假设我们点击bn1开启协程后，立即点击bn2取消协程（此时协程仍然在Thread.sleep期间），那么输出为

D/abcde: onCreate: false
D/abcde: onCreate: DefaultDispatcher-worker-2,Top Scope

可以看到，协程的isActive的值变为false，但是协程仍然会执行（虽然之后无法通过scope再去启动新的协程）。

上面的例子中，已经调用了scope.cancel，但是当前协程仍然还在运行，说明协程的真正取消需要协程内部的配合，其中一个方法就是调用ensureActive()函数，ensureActive的作用大致上相当于：

if (!isActive) {
    throw CancellationException()
}

我们修改下上面的例子：

val scope = CoroutineScope(Job() + Dispatchers.IO + CoroutineName("Top Scope"))

bn1.setOnClickListener {
	scope.launch {
        Thread.sleep(2000)
        Log.d(TAG, "onCreate: $isActive")
        // 检查协程是否取消
        ensureActive()
        Log.d(TAG, "onCreate: ${threadName()},${coroutineContext[CoroutineName]?.name}")
	}
}

bn2.setOnClickListener {
	scope.cancel()
}

我们点击bn1开启协程后，立即点击bn2取消协程（此时协程仍然在Thread.sleep期间），那么输出为

D/abcde: onCreate: false

可以看到，当前协程内部的ensureActive()函数配合外部的cancel操作，成功地将协程取消了。

当然，也可以通过其它的方式在协程内部进行协作式地取消操作。

协程挂起

外部对协程cancel之后，运行的协程被suspend的时候，协程也会被取消。

对上面的例子改造一下：

val scope = CoroutineScope(Job() + Dispatchers.IO + CoroutineName("Top Scope"))

bn1.setOnClickListener {
    scope.launch {
        Thread.sleep(2000)
        Log.d(TAG, "onCreate: $isActive")
        withContext(Dispatchers.Main) {
            Log.d(TAG, 
                  "onCreate: ${threadName()},${coroutineContext[CoroutineName]?.name}")
        }
    }    
}

bn2.setOnClickListener {
	scope.cancel()
}

假如我们只点击bn1开启协程，但是不点击bn2去取消协程，那么输出为

D/abcde: onCreate: true
D/abcde: onCreate: main,Top Scope

假设我们点击bn1开启协程后，立即点击bn2取消协程（此时协程仍然在Thread.sleep期间），那么输出为

D/abcde: onCreate: false

可以看出，withContext在suspend当前协程的时候，协程被取消了。

kotlinx.coroutines中的所有suspend函数都是可取消的(cancellable)，例如withContext and delay(上面的例子中，不使用withContext，使用delay函数也是可以实现协程的取消的)。如果协程中调用了这些挂起函数，就不需要做任何其它的额外工作。

异常的处理

对于协程中的异常，可以使用try...catch...进行捕获，也可以使用CoroutineExceptionHandler。

CoroutineExceptionHandler是CoroutineContext中的一种

协程中使用try...catch...捕获异常：

class LoginViewModel(
    private val loginRepository: LoginRepository
) : ViewModel() {

    fun login(username: String, token: String) {
        viewModelScope.launch {
            try {
                loginRepository.login(username, token)
                // Notify view user logged in successfully
            } catch (error: Throwable) {
                // Notify view login attempt failed
            }
        }
    }
}

其它挂起函数

coroutineScope

挂起函数coroutineScope：创建一个CoroutineScope，并且在这个scope里面调用特定的suspend block，创建的CoroutineScope继承外部scope的CoroutineContext（CoroutineContext中的Job会被重写）。

这个函数为parallel decomposition而设计，当这个scope的任何子协程fail，这个scope里面其它的子协程也会fail，这个scope也fail了（感觉有点结构化并发的感觉）。

当使用coroutineScope的时候，外部的协程会被挂起，直到coroutineScope里面的代码和scope里面的协程运行结束的时候，挂起函数coroutineScope的外部协程就会恢复执行。

一个例子：

    GlobalScope.launch(Dispatchers.Main) {
        fetchTwoDocs()
        Log.d(TAG, "Under fetchTwoDocs()")
    }

    suspend fun fetchTwoDocs() {
        coroutineScope {
            Log.d(TAG, "fetchTwoDocs: ${threadName()}")
            val deferredOne = async {
                Log.d(TAG, "async1 start: ${threadName()}")
                fetchDoc1()
                Log.d(TAG, "async1 end: ${threadName()}")
            }
            val deferredTwo = async {
                Log.d(TAG, "async2: start:${threadName()}")
                fetchDoc2()
                Log.d(TAG, "async2 end: ${threadName()}")
            }
            deferredOne.await()
            deferredTwo.await()
        }
    }

    suspend fun fetchDoc1() = withContext(Dispatchers.IO) {
        Thread.sleep(2000L)
    }

    suspend fun fetchDoc2() = withContext(Dispatchers.IO) {
        Thread.sleep(1000L)
    }

D/abcde: fetchTwoDocs: main
D/abcde: async1 start: main
D/abcde: async2: start:main
D/abcde: async2 end: main
D/abcde: async1 end: main
D/abcde: Under fetchTwoDocs()

几个关注点：

  1. Under fetchTwoDocs()在fetchTwoDocs执行完毕后才输出
  2. coroutineScope里面的代码在主线程运行
  3. async的代码运行在main线程中，因为coroutineScope创建的scope会继承外部的GlobalScope.launch的CoroutineContext。

上面的代码即使不调用deferredOne.await()、deferredTwo.await()，也是一样的执行和输出结果。

suspendCoroutine

/**
 * Obtains the current continuation instance inside suspend functions and suspends
 * the currently running coroutine.
 *
 * In this function both [Continuation.resume] and [Continuation.resumeWithException] can be used either synchronously in
 * the same stack-frame where the suspension function is run or asynchronously later in the same thread or
 * from a different thread of execution. Subsequent invocation of any resume function will produce an [IllegalStateException].
 */
@SinceKotlin("1.3")
@InlineOnly
public suspend inline fun <T> suspendCoroutine(crossinline block: (Continuation<T>) -> Unit): T {
    contract { callsInPlace(block, InvocationKind.EXACTLY_ONCE) }
    return suspendCoroutineUninterceptedOrReturn { c: Continuation<T> ->
        val safe = SafeContinuation(c.intercepted())
        block(safe)
        safe.getOrThrow()
    }
}

suspendCoroutine是一个主动挂起协程的行为，它会给你一个Continuation，让你决定什么时候去恢复协程的执行。

参考

  1. Kotlin coroutines on Android | Android Developers。
上一篇：使用ResNet101作为预训练模型训练Faster-RCNN-TensorFlow-Python3-master


下一篇：订单峰值激增 230%，Serverless 如何为世纪联华降本超 40%？｜双11 云原生实践