投稿
  1. 首页
  2. app

Kotlin协程的使用

code • 2022-5-21 • app • 阅读 44

Kotlin协程的使用,第1张

前言

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

协程概述

一、协程是什么？

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

关于协程作用域：协程必须运行在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（返回Job）与async（返回Deferred）的区别：

launch启动的协程没有返回结果；async启动的协程有返回结果，该结果可以通过Deferred的await方法获取。launch启动的协程有异常会立即抛出；async启动的协程的异常不会立即抛出，会等到调用Deferred::await的时候才将异常抛出。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类的部分方法：

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

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

Job的生命周期：New、 Active、Completing、 Completed、 Cancelling、Cancelled。虽然我们无法访问状态本身，但是我们可以调用Job的isActive、isCancelled、isCompleted方法

当协程是Active的时候，协程failure或者调用Job.cancel方法将会让Job进入Cancelling状态（isActive = false, isCancelled = true）。一旦所有的子协程完成了它们的工作，外部协程将会进入Cancelled状态，此时isCompleted = true。

CoroutineContext

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

Job：控制协程的生命周期。

CoroutineDispatcher：将工作分派到适当的线程。

默认是Dispatchers.Default

CoroutineName：协程的名称，可用于调试。

默认是“coroutine”

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。

在协程中访问元素

通过CoroutineScope的launch或者async可以启动一个协程：

public fun CoroutineScope.launch(...)
public fun <T> CoroutineScope.async(...)

在CoroutineScope中有协程上下文元素：

public interface CoroutineScope {
    public val coroutineContext: CoroutineContext
}

因此在协程中可以直接访问到协程上下文元素：

scope.launch {
    coroutineContext
	...
}

也有一些方便的扩展函数，例如：

public val CoroutineScope.isActive: Boolean
    get() = coroutineContext[Job]?.isActive ?: true

这样在协程中就可以直接获取Job的状态：

scope.launch {
    isActive
	...
}
协程的层级关系

每一个协程都有一个parent，这个parent可以是CoroutineScope，也可以是另一个协程。

创建新协程的CoroutineContext计算公式：

Defaults + inherited CoroutineContext + arguments + Job()

若有重复的协程上下文元素，则+号右边的元素会覆盖+号左边的元素。公式各部分含义如下：

Defaults：例如默认的Dispatchers.Default（CoroutineDispatcher）和“coroutine”（CoroutineName）inherited CoroutineContext：从 CoroutineScope或协程 继承的协程上下文元素arguments：通过协程构造器launch或者async传入的元素Job()：新协程总是可以获取一个新的Job实例 避免使用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。

协程的取消

通过 Job 取消单个协程，但不影响该作用域下的其它协程：

// assume we have a scope defined for this layer of the app
val job1 = scope.launch { … }
val job2 = scope.launch { … }
// First coroutine will be cancelled and the other one won’t be affected
job1.cancel()

通过协程作用域 CoroutineScope 可以取消该作用域下的所有协程（一旦取消协程作用域，将不能使用该作用域去启动新的协程）：

// assume we have a scope defined for this layer of the app
val job1 = scope.launch { … }
val job2 = scope.launch { … }
scope.cancel()

取消一个协程可以使用 cancel 方法：

# Job.kt
public fun cancel(): Unit = cancel(null)

如果想为该取消提供更多的信息或理由，可以调用下面的方法，自己传入一个CancellationException：

# Job.kt
public fun cancel(cause: CancellationException? = null)

没有参数的 cancel 方法，其实也是调用了这个带有 CancellationException 参数的 cancel 方法。如果使用没有参数的 cancel 方法，那么会使用系统提供的默认的 CancellationException ：

# Job.kt
public fun cancel(): Unit = cancel(null)

public fun cancel(cause: CancellationException? = null)

# JobSupport.kt
public override fun cancel(cause: CancellationException?) {
    cancelInternal(cause ?: defaultCancellationException())
}

internal inline fun defaultCancellationException(message: String? = null, cause: Throwable? = null) =
	JobCancellationException(message ?: cancellationExceptionMessage(), cause, this)

官方文档的原话：

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.

可以得出：

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

协程内部可以通过抛出 CancellationException 异常来处理取消

主动检查

举个例子：

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（内部调用了job.cancel）的时候，协程会进入Cancelling 状态，当协程内所有的工作都完成了，协程会进入 Cancelled 状态。

上面的例子中，已经调用了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函数也是可以实现协程的取消的)。

这是因为协程调度器 CoroutineDispatcher 在继续正常执行之前会检查协程对应的 Job 的状态，如果 Job 被取消了，那么 CoroutineDispatcher 会终止正常的执行，并且调用相应的 cancellation handlers，下面是一个例子：

var job: Job? = null

// 启动协程
binding.start.setOnClickListener {
    job = scope.launch {
        withContext(Dispatchers.IO){
            Thread.sleep(1000)
            Log.d(TAG, "1")
        }
        Log.d(TAG, "2")
    }
}

// 取消协程
binding.cancel.setOnClickListener {
    job?.cancel()
}

先点击按钮启动协程，在协程的 Thread.sleep 执行期间，点击按钮取消协程，那么输出为：

D/abcde: 1

另外，使用yield()函数也可以让协程响应取消，后面「其它挂起函数」一节会介绍原因。

Job.join vs Deferred.await Job.join

Job.join可以让当前协程挂起，直到另外一个协程执行完毕为止。

Job.join 和 Job.cancel：

如果调用了 Job.cancel，然后调用 Job.join，那么当前协程会挂起，直到 Job 对应的协程执行完毕。如果调用了 Job.join，然后调用 Job.cancel，那么 Job.cancel 的调用不会产生任何效果，因为当 Job.join 执行完毕后，Job 对应的协程已经完成了，这时候 Job.cancel 的执行不会产生任何影响。

例子一：

var job : Job? = null

// 先点击start按钮
binding.start.setOnClickListener {
    job = scope.launch { 
        Thread.sleep(1000)
        Log.d(TAG, "协程1")
    }
}

// 然后点击cancel按钮
binding.cancel.setOnClickListener {
    scope.launch {
        job?.cancel()
        job?.join()
        Thread.sleep(1000)
        Log.d(TAG, "协程2")
    }
}

D/abcde: 协程1
D/abcde: 协程2

例子二（调换了cancel和join的执行顺序）：

var job : Job? = null
binding.start.setOnClickListener {
    job = scope.launch { 
        Thread.sleep(1000)
        Log.d(TAG, "协程1")
    }
}

binding.cancel.setOnClickListener {
    scope.launch {
        job?.join()
        job?.cancel()
        Thread.sleep(1000)
        Log.d(TAG, "协程2")
    }
}

D/abcde: 协程1
D/abcde: 协程2
Deferred.await

通过async启动一个协程会返回一个Deferred，通过Deferred的await函数可以获取协程的执行结果。

如果在调用了deferred.cancel后再调用deferred.await，那么deferred.await的调用会抛出一个JobCancellationException异常，如下：

var deferred : Deferred<Int>? = null

binding.start.setOnClickListener {
    // 协程1
	deferred = scope.async {
        Thread.sleep(1000)
        Log.d(TAG, "print 1")
        1    
	}
}

binding.cancel.setOnClickListener {
    // 协程2
    scope.launch {
        deferred?.cancel()
        deferred?.await()
        Log.d(TAG, "after cancel")
    }
}
如果在协程1执行完毕之前，启动协程2，对协程cancel然后await，这时await的调用会抛出JobCancellationException异常，该异常会结束协程2的运行。 
输出：
D/abcde: print 1

抛出异常的原因：await的作用是等待协程1的计算结果，由于调用deferred.cancel使得协程1被取消了，因此调用deferred.await时协程1无法计算出结果，因此会抛出异常。

如果在协程1执行完毕之后，启动协程2，这时候由于协程1已执行完毕，所以deferred.cancel没有任何效果，之后再调用deferred.await也不会抛出异常，一切运行正常。 
输出：
D/abcde: print 1
D/abcde: after cancel

将上面代码中cancel和await的调用顺序对调，即：

var deferred : Deferred<Int>? = null

binding.start.setOnClickListener {
    // 协程1
	deferred = scope.async {
        Thread.sleep(1000)
        Log.d(TAG, "print 1")
        1    
	}
}

binding.cancel.setOnClickListener {
    // 协程2
    scope.launch {
        deferred?.await()
        deferred?.cancel()
        Log.d(TAG, "after cancel")
    }
}

输出：
D/abcde: print 1
D/abcde: after cancel

由于调用await，协程2会等待协程1执行完毕，当调用cancel的时候，由于协程1已经执行完毕了，这时候该cancel函数不会产生任何的效果。

清理资源

当协程取消的时候，如果想协程可以响应取消，并且清理资源，有两种办法：

手动检查协程是否 Active 状态，从而控制协程的执行使用 Try Catch Finally 语句块

一、手动检查协程是否 Active 状态，从而控制协程的执行

while (i < 5 && isActive) {
    // print a message twice a second
    if (…) {
        println(“Hello ${i++}”)
        nextPrintTime += 500L
    }
}
// the coroutine work is completed so we can cleanup
println(“Clean up!”)

当协程不是 Active 状态时，退出while循环，做一些资源清理的动作，然后结束协程。

二、使用 Try Catch Finally 语句块

suspend fun work(){
    val startTime = System.currentTimeMillis()
    var nextPrintTime = startTime
    var i = 0
    while (i < 5) {
        yield()
        // print a message twice a second
        if (System.currentTimeMillis() >= nextPrintTime) {
            println("Hello ${i++}")
            nextPrintTime += 500L
        }
    }
}
fun main(args: Array<String>) = runBlocking<Unit> {
   val job = launch (Dispatchers.Default) {
        try {
        	work()
        } catch (e: CancellationException){
            println("Work cancelled!")
        } finally {
            println("Clean up!")
        }
    }
    delay(1000L)
    println("Cancel!")
    job.cancel()
    println("Done!")
}

work函数检查到协程不是 Active 的时候，会抛出 CancellationException 异常，可以使用 Try Catch 语句块捕获该异常，然后在 Finally 块中做资源清理的动作。

注意：当协程是 Cancelling 状态的时候，协程就无法被挂起了，如果此时再调用挂起函数尝试挂起协程，那么协程的执行会结束。这意味着如果协程被取消了，那么协程的清理资源的代码中不能调用挂起函数将协程挂起。如下

fun main(args: Array<String>) = runBlocking<Unit> {
    val job = launch {
        try {
            work()
        } catch (e: CancellationException) {
            println("Work cancelled!")
        } finally {
            delay(2000L) // or some other suspend fun
            println("Cleanup done!") // 没有输出
        }
    }
    delay(1000L)
    println("Cancel!")
    job.cancel()
    println("Done!")
}

Hello 0
Hello 1
Cancel!
Done!
Work cancelled!

但是有一种特殊的方法，可以在协程取消的时候仍然可以执行将协程挂起的代码，我们需要将「使协程挂起的代码」放在 NonCancellable CoroutineContext 下面执行，具体做法是将 Finally 块的语句修改为：

    withContext(NonCancellable) {
        delay(2000L) // or some other suspend fun
        println("Cleanup done!") // 成功输出
    }

Hello 0
Hello 1
Cancel!
Done!
Work cancelled!
Cleanup done!
异常的处理

对于协程中的异常，可以使用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()

几个关注点：

Under fetchTwoDocs()在fetchTwoDocs执行完毕后才输出coroutineScope里面的代码在主线程运行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，让你决定什么时候去恢复协程的执行。

suspendCancellableCoroutine

与 suspendCoroutine 一样，suspendCancellableCoroutine 也可以将协程挂起，它们的不同之处在于，suspendCancellableCoroutine 对协程的取消做了一些处理与支持。

当协程挂起的时候，如果协程的 Job 被取消了，那么协程将无法成功地 resume，并且 suspendCancellableCoroutine 函数会抛出一个 CancellationException 。

suspendCancellableCoroutine 函数的定义如下：

public suspend inline fun <T> suspendCancellableCoroutine(
    crossinline block: (CancellableContinuation<T>) -> Unit
): T =
    suspendCoroutineUninterceptedOrReturn { uCont ->
        val cancellable = CancellableContinuationImpl(uCont.intercepted(), resumeMode = MODE_CANCELLABLE)
        /*
         * For non-atomic cancellation we setup parent-child relationship immediately
         * in case when `block` blocks the current thread (e.g. Rx2 with trampoline scheduler), but
         * properly supports cancellation.
         */
        cancellable.initCancellability()
        block(cancellable)
        cancellable.getResult()
    }

可以看出，它为传入的 Lambda 提供了一个 CancellableContinuation 参数，CancellableContinuation 提供了 invokeOnCancellation 函数，当一个挂起的协程被取消的时候，invokeOnCancellation 函数中的代码会被执行：

     suspendCancellableCoroutine { continuation ->
         val resource = openResource() // Opens some resource
         continuation.invokeOnCancellation {
             resource.close() // Ensures the resource is closed on cancellation
         }
         // ...
     }

使用 CancellableContinuation 恢复协程的时候，即调用它的 resume 函数的时候，可以额外传入一个 onCancellation 参数：

suspendCancellableCoroutine { continuation ->
    val callback = object : Callback { // Implementation of some callback interface
        // A callback provides a reference to some closeable resource
        override fun onCompleted(resource: T) {
            // Resume coroutine with a value provided by the callback and ensure the 
            // resource is closed in case when the coroutine is cancelled before the 
            // caller gets a reference to the resource.
            continuation.resume(resource) { // 额外传入的onCancellation参数，是一个Lambda
                resource.close() // Close the resource on cancellation
            }
        }
    }
    // ...
}

如果在续体的 resume 执行之前，协程就被取消了，那么调用者将无法获取 resource 的引用，调用者就无法对 resource 资源进行关闭。这时候 onCancellation 参数就派上用场了，它可以在上面的情况发生时，对 resource 资源进行关闭。

yield

如果想让当前的协程让出线程，可以使用yield()这个挂起函数。

之前也提到过，yield函数可以响应协程的取消，这是因为yield函数中做的第一个 *** 作就是检查协程的状态，当协程不是 Active 时就会抛出CancellationException：

public suspend fun yield(): Unit = suspendCoroutineUninterceptedOrReturn sc@ { uCont ->
    val context = uCont.context
    context.checkCompletion()
    ...                                                                          
}          

internal fun CoroutineContext.checkCompletion() {
    val job = get(Job)
    if (job != null && !job.isActive) throw job.getCancellationException()
}
参考

Kotlin coroutines on Android | Android Developers。

Coroutines: first things first. Cancellation and Exceptions in… | by Manuel Vivo | Android Developers | Medium。

欢迎分享，转载请注明来源：内存溢出

原文地址: http://outofmemory.cn/web/989770.html

赞 (0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
code code 管理员组
0 0
上一篇 2022-05-21
下一篇 2022-05-21

发表评论

登录后才能评论

评论列表（0条）

保存