（译）Android中的Kotlin协程-基础

2020-02-27 本文已影响0人剑舞潇湘

如果英文较好，建议直接阅读原文

译文

什么是协程

基本上，coroutines是轻量级线程，它使得我们可以用串行的方式写出异步的、非阻塞的代码。

Android中如何导入Kotlin协程

根据Kotlin Coroutines Github repo，我们需要导入kotlinx-coroutines-core和kotlinx-coroutines-android（类似于RxJava的io.reactivex.rxjava2:rxandroid，该库支持Android主线程，同时保证未捕获的异常可以在应用崩溃前输出日志）。如果项目里使用了RxJava，可以导入kotlinx-coroutines-rx2来同时使用RxJava和协程，这个库帮助将RxJava代码转为协程。

添加如下代码导入

implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.2'
implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.3.2'
implementation "org.jetbrains.kotlinx:kotlinx-coroutines-rx2:1.3.2"

记得添加最新的Kotlin版本到根build.gradle：

buildscript {
    ext.kotlin_version = '1.3.50'
    repositories {
        jcenter()
        ...
    }
    ...
}

OK，准备工作已就绪，让我们开始吧～

内容目录

Suspending functions
Coroutine scope
(1) CoroutineScope
(2) MainScope
(3) GlobalScope
Coroutine context
(1) Dispatchers
(2) CoroutineExceptionHandler
(3) Job
— (3.1) Parent-child hierarchies
— (3.2) SupervisorJob v.s. Job
Coroutine builder
(1) launch
(2) async
Coroutine body

协程基础

先看看协程长啥样：

CoroutineScope(Dispatchers.Main + Job()).launch {
  val user = fetchUser() // A suspending function running in the I/O thread.
  updateUser(user) // Updates UI in the main thread.
}

private suspend fun fetchUser(): User = withContext(Dispatchers.IO) {
  // Fetches the data from server and returns user data.
}

这段代码在后台线程拉取服务器数据，然后回到主线程更新UI.

1. Suspending functions

suspending functions是Kotlin协程中的特殊方法，用关键字suspend定义。Suspending functions可以中断（suspend）当前协程的执行，这意味着它一直等待，直到suspending functions恢复（resume）。因为这篇博客关注协程的基本概念， we’ll discuss more detail of suspending functions in this post.

我们回过头来看看上面的代码，它可以分为4个部分：

suspend functions.png

2. CoroutineScope（协程作用域）

Defines a scope for new coroutines. Every coroutine builder is an extension on CoroutineScope and inherits its coroutineContext to automatically propagate both context elements and cancellation.

为一系列新协程定义作用域. 每个协程构建器都是CoroutineScope的拓展，继承其coroutineContext以自动地（向下）传递上下文对象和取消。

所有的协程都在协程作用域里运行，并接受一个CoroutineContext（协程上下文，后文详述）作为参数。有几个作用域我们可以使用：

(1) CoroutineScope

用自定义的CoroutineContext（协程上下文）创建作用域。例如，根据我们的需要，指定线程、父job和异常处理器（the thread, parent job and exception handler）：

CoroutineScope(Dispatchers.Main + job + exceptionHandler).launch {
    ...
}

(2) MainScope

为UI组件创建一个主作用域。它使用SupervisorJob()，在主线程运行，这意味着如果它的某个子任务（child job）失败了，不会影响其他子任务。

public fun MainScope(): CoroutineScope = ContextScope(SupervisorJob() + Dispatchers.Main)

(3) GlobalScope

这个作用域不跟任何任务（job）绑定。它用来启动顶级协程，这些协程可以运行在应用的整个生命周期，且永远不能取消。

3. CoroutineContext（协程上下文）

协程总是运行在某个CoroutineContext类型的context中。CoroutineContext是一系列元素，用来指定线程策略、异常处理器、控制协程生命周期等。可以用+操作符将这些元素组合起来。

有3种最重要的协程上下文：Dispatchers，CoroutineExceptionHandler，Job

(1) Dispatchers

指定协程在哪个线程执行。协程可以随时用withContext()切换线程。

Dispatchers.Default

使用共享的后台线程缓存池。默认情况下，它使用的最大线程数等于CPU内核数，但至少2个。这个线程看起来会像是Thread[DefaultDispatcher-worker-2,5,main].

Dispatchers.IO

跟Dispatchers.Default共享线程，但它数量受kotlinx.coroutines.io.parallelism限制，默认最多是64个线程或CPU内核数（其中的大值）。跟Dispatchers.Default一样，线程看起来像Thread[DefaultDispatcher-worker-1,5,main].

Dispatchers.Main

等效于主线程。线程看起来像Thread[main,5,main].

Dispatchers.Unconfined

未指定特定线程的协程分发器。协程在当前线程执行，并让协程恢复到对应的suspending function用过的任意线程上。

CoroutineScope(Dispatchers.Unconfined).launch {
    // Writes code here running on Main thread.
    
    delay(1_000)
    // Writes code here running on `kotlinx.coroutines.DefaultExecutor`.
    
    withContext(Dispatchers.IO) { ... }
    // Writes code running on I/O thread.
    
    withContext(Dispatchers.Main) { ... }
    // Writes code running on Main thread.
}

(2) CoroutineExceptionHandler

处理未捕获的异常。

Normally, uncaught exceptions can only result from coroutines created using the launch builder. A coroutine that was created using async always catches all its exceptions and represents them in the resulting Deferred object.

例子1：不能通过外层try-catch捕获IOException()。不能用try-catch包围整个协程作用域，否则应用还是会崩溃。

try {
  CoroutineScope(Dispatchers.Main).launch {
    doSomething()
  }
} catch (e: IOException) {
  //  无法捕获IOException()
  Log.d("demo", "try-catch: $e")
}

private suspend fun doSomething() {
  delay(1_000)
  throw IOException()
}

例子2：用CoroutineExceptionHandler捕获IOException()。除CancellationException外的其他异常，如IOException()，将传递给CoroutineExceptionHandler。

// Handles coroutine exception here.
val handler = CoroutineExceptionHandler { _, throwable ->
  Log.d("demo", "handler: $throwable") // Prints "handler: java.io.IOException"
}

CoroutineScope(Dispatchers.Main + handler).launch {
  doSomething()
}

private suspend fun doSomething() {
  delay(1_000)
  throw IOException()
}

例子3：CancellationException()会被忽略。

如果协程抛出CancellationException，它将会被忽略（因为这是取消运行中的协程的预期机制，所以该异常不会传递给CoroutineExceptionHandler）（译注：不会导致崩溃）

// Handles coroutine exception here.
val handler = CoroutineExceptionHandler { _, throwable ->
  // Won't print the log because the exception is "CancellationException()".
  Log.d("demo", "handler: $throwable")
}

CoroutineScope(Dispatchers.Main + handler).launch {
  doSomething()
}

private suspend fun doSomething() {
  delay(1_000)
  throw CancellationException()
}

例子4：用invokeOnCompletion可以获取所有异常信息。

CancellationException不会传递给CoroutineExceptionHandler，但当该异常发生时，如果我们想打印出某些信息，可以使用invokeOnCompletion来获取。

val job = CoroutineScope(Dispatchers.Main).launch {
  doSomething()
}

job.invokeOnCompletion {
    val error = it ?: return@invokeOnCompletion
    // Prints "invokeOnCompletion: java.util.concurrent.CancellationException".
    Log.d("demo", "invokeOnCompletion: $error")
  }
}

private suspend fun doSomething() {
  delay(1_000)
  throw CancellationException()
}

(3) Job

控制协程的生命周期。一个协程有如下状态：

job状态.png

查询job的当前状态很简单，用Job.isActive。

状态流图是：

job状态流图.png

协程工作时job是active态的
job发生异常时将会变成cancelling. 一个job可以随时用cancel方法取消，这个强制使它立刻变为cancelling态
当job工作完成时，会变成cancelled态
父job会维持在completing或cancelling态直到所有子job完成。注意completing是一种内部状态，对外部来说，completing态的job仍然是active的。

(3.1) Parent-child hierarchies（父-子层级）

弄明白状态后，我门还必须知道父-子层级是如何工作的。假设我们写了如下代码：

val parentJob = Job()
val childJob1 = CoroutineScope(parentJob).launch {
    val childJob2 = launch { ... }
    val childJob3 = launch { ... }
}

则其父子层级会长这样：

job父子层级.png

我们可以改变父job，像这样：

val parentJob1 = Job()
val parentJob2 = Job()
val childJob1 = CoroutineScope(parentJob1).launch {
    val childJob2 = launch { ... }
    val childJob3 = launch(parentJob2) { ... }
}

则父子层级会长这样：

job父子层级2.png

基于以上知识，我们需要知道如下一些重要概念：

父job取消将立即导致所有子job取消

val parentJob = Job()
CoroutineScope(Dispatchers.Main + parentJob).launch {
    val childJob = launch {
        delay(5_000)
        
        // This function won't be executed because its parentJob is 
        // already cancelled after 1 sec. 
        canNOTBeExcecuted()
    }
    launch {
        delay(1_000)
        parentJob.cancel() // Cancels parent job after 1 sec.
    }
}

当某个子job因为除CancellationException外的异常而失败或取消时，会立刻导致所有父job和其他子job取消。但如果是CancellationException，则除该job的子job外的其他jobs不会收到影响。

例子1：如果抛出CancellationException，只有childJob1下的job被取消。

val parentJob = Job()
CoroutineScope(Dispatchers.Main + parentJob).launch {
  val childJob1 = launch {
    val childOfChildJob1 = launch {
      delay(2_000)
      // This function won't be executed since childJob1 is cancelled.
      canNOTBeExecuted()
    }
    delay(1_000)
    
    // Cancel childJob1.
    cancel()
  }

  val childJob2 = launch {
    delay(2_000)
    canDoSomethinghHere()
  }

  delay(3_000)
  canDoSomethinghHere()
}

例子2：如果某个子job抛出IOException，则所有关联job都会被取消

val parentJob = Job()
val handler = CoroutineExceptionHandler { _, throwable ->
  Log.d("demo", "handler: $throwable") // Prints "handler: java.io.IOException"
}

CoroutineScope(Dispatchers.Main + parentJob + handler).launch {
  val childJob1 = launch {
    delay(1_000)
    // Throws any exception "other than CancellationException" after 1 sec.
    throw IOException() 
  }

  val childJob2 = launch {
    delay(2_000)
    // The other child job: this function won't be executed.
    canNOTBExecuted()
  }

  delay(3_000)
  // Parent job: this function won't be executed.
  canNOTBExecuted()
}

cancelChildren(): 父job可以取消它的所有子job（递归到它们的子job）而被取消自己。注意：如果一个job已取消，则它不能再作为父job运行协程了。

如果我们用Job.cancel()，父job将会变成cancelled(当前是Cancelling)，当其所有子job都cancelled后，父job会成为cancelled态。

val parentJob = Job()
val childJob = CoroutineScope(Dispatchers.Main + parentJob).launch {
  delay(1_000)
  
  // This function won't be executed because its parent is cancelled.
  canNOTBeExecuted()
}

parentJob.cancel()

// Prints "JobImpl{Cancelling}@199d143", parent job status becomes "cancelling".
// And will be "cancelled" after all the child job is cancelled.
Log.d("demo", "$parentJob")

而如果我们用Job.cancelChildren()，父job将会变为Active态，我们仍然可以用它来运行其他协程。

val parentJob = Job()
val childJob = CoroutineScope(Dispatchers.Main + parentJob).launch {
  delay(1_000)
  
  // This function won't be executed because its parent job is cancelled.
  canNOTBeExecuted()
}

// Only children are cancelled, the parent job won't be cancelled.
parentJob.cancelChildren()

// Prints "JobImpl{Active}@199d143", parent job is still active.
Log.d("demo", "$parentJob")

val childJob2 = CoroutineScope(Dispatchers.Main + parentJob).launch {
  delay(1_000)
  
  // Since the parent job is still active, we could use it to run child job 2.
  canDoSomethingHere()
}

(3.2) SupervisorJob v.s. Job

supervisor job的子job可以独立失败，而不影响其他子job。

正如前文提到的，如果我们用Job()作为父job，当某个子job失败时将会导致所有子job取消。

val parentJob = Job()
val handler = CoroutineExceptionHandler { _, _ -> }
val scope = CoroutineScope(Dispatchers.Default + parentJob + handler)
val childJob1 = scope.launch {
    delay(1_000)
    // ChildJob1 fails with the IOException().
    throw IOException()
}

val childJob2 = scope.launch {
    delay(2_000)
    // This line won't be executed due to childJob1 failure.
    canNOTBeExecuted()
}

如果我们使用SupervisorJob()作为父job，则其中一个子job取消时不会影响其他子jobs。

val parentJob = SupervisorJob()
val handler = CoroutineExceptionHandler { _, _ -> }
val scope = CoroutineScope(Dispatchers.Default + parentJob + handler)
val childJob1 = scope.launch {
    delay(1_000)
    // ChildJob1 fails with the IOException().
    throw IOException()
}

val childJob2 = scope.launch {
    delay(2_000)
    // Since we use SupervisorJob() as parent job, the failure of
    // childJob1 won't affect other child jobs. This function will be 
    // executed.
    canDoSomethinghHere()
}

4. Coroutines Builder（协程构建器）

(1) launch

启动一个新协程，不会阻塞当前线程，返回一个指向当前协程的Job引用。

(2) async and await

async协程构建器是CoroutineScope的拓展方法。它创建一个协程，并以Deferred实现来返回它的未来结果，这是一个非阻塞的可取消future——一个带结果的Job。

Async协程搭配await使用：不阻塞当前线程的前提下持续等待结果，并在可延迟的任务完成后恢复（resume），返回结果，或者如果deferred被取消了，抛出相应的异常。

下列代码展示了两个suspending functions的串行调用。在fetchDataFromServerOne()和fetchDataFromServerTwo()中，我们做了一些耗时任务，分别耗时1秒。在launch构建器里调用它们，会发现最终的耗时是它们的和：2秒。

override fun onCreate(savedInstanceState: Bundle?) {
  ...

  val scope = MainScope()
  scope.launch {
    val time = measureTimeMillis {
      val one = fetchDataFromServerOne()
      val two = fetchDataFromServerTwo()
      Log.d("demo", "The sum is ${one + two}")
    }
    Log.d("demo", "Completed in $time ms")
  }
}

private suspend fun fetchDataFromServerOne(): Int {
  Log.d("demo", "fetchDataFromServerOne()")
  delay(1_000)
  return 1
}
  
private suspend fun fetchDataFromServerTwo(): Int {
  Log.d("demo", "fetchDataFromServerTwo()")
  delay(1_000)
  return 2
}

日志是：

2019-12-09 00:00:34.547 D/demo: fetchDataFromServerOne()
2019-12-09 00:00:35.553 D/demo: fetchDataFromServerTwo()
2019-12-09 00:00:36.555 D/demo: The sum is 3
2019-12-09 00:00:36.555 D/demo: Completed in 2008 ms

耗时是两个suspending functions延时的和。该协程在fetchDataFromServerOne()结束前会中断（suspend），然后执行fetchDataFromServerTwo()。

如果我们想同时运行两个方法以减少耗时呢？Async闪亮登场！Async和launch很像。它启动一个可以和其他协程同时运行的新协程，返回Deferred引用——一个带返回值的Job。

public interface Deferred<out T> : Job {
  public suspend fun await(): T
  ...
}

在Deferred上调用await()获取结果，例如：

override fun onCreate(savedInstanceState: Bundle?) {
  ...
  
  val scope = MainScope()
  scope.launch {
    val time = measureTimeMillis {
      val one = async { fetchDataFromServerOne() }
      val two = async { fetchDataFromServerTwo() }
      Log.d("demo", "The sum is ${one.await() + two.await()}")
    }
    
    // Function one and two will run asynchrously,
    // so the time cost will be around 1 sec only. 
    Log.d("demo", "Completed in $time ms")
  }
}

private suspend fun fetchDataFromServerOne(): Int {
  Log.d("demo", "fetchDataFromServerOne()")
  delay(1_000)
  return 1
}

private suspend fun fetchDataFromServerTwo(): Int {
  Log.d("demo", "fetchDataFromServerTwo()")
  Thread.sleep(1_000)
  return 2
}