盒子
盒子
文章目录
  1. CoroutineScope
  2. Job
    1. invokeOnCompletion
    2. join
  3. 项目

Kotlin协程实现原理:CoroutineScope&Job

今天我们来聊聊Kotlin的协程Coroutine。

如果你还没有接触过协程,推荐你先阅读这篇入门级文章What? 你还不知道Kotlin Coroutine?

如果你已经接触过协程,但对协程的原理存在疑惑,那么在阅读本篇文章之前推荐你先阅读下面的文章,这样能让你更全面更顺畅的理解这篇文章。

Kotlin协程实现原理:Suspend&CoroutineContext

如果你已经接触过协程,相信你都有过以下几个疑问:

  1. 协程到底是个什么东西?
  2. 协程的suspend有什么作用,工作原理是怎样的?
  3. 协程中的一些关键名称(例如:Job、Coroutine、Dispatcher、4. CoroutineContext与CoroutineScope)它们之间到底是怎么样的关系?
  4. 协程的所谓非阻塞式挂起与恢复又是什么?
  5. 协程的内部实现原理是怎么样的?

接下来的一些文章试着来分析一下这些疑问,也欢迎大家一起加入来讨论。

CoroutineScope

CoroutineScope是什么?如果你觉得陌生,那么GlobalScope、lifecycleScope与viewModelScope相信就很熟悉了吧(当然这个是针对于Android开发者)。它们都实现了CoroutineScope接口。

1
2
3
4
5
6
7
8
9
10
public interface CoroutineScope {
/**
* The context of this scope.
* Context is encapsulated by the scope and used for implementation of coroutine builders that are extensions on the scope.
* Accessing this property in general code is not recommended for any purposes except accessing the [Job] instance for advanced usages.
*
* By convention, should contain an instance of a [job][Job] to enforce structured concurrency.
*/
public val coroutineContext: CoroutineContext
}

CoroutineScope中只包含一个待实现的变量CoroutineContext,至于CoroutineContext之前的文章已经分析了它的内部结构,这里就不再累赘了。

通过它的结构,我们可以认为它是提供CoroutineContext的容器,保证CoroutineContext能在整个协程运行中传递下去,约束CoroutineContext的作用边界。

例如,在Android中使用协程来请求数据,当接口还没有请求完成时Activity就已经退出了,这时如果不停止正在运行的协程将会造成不可预期的后果。所以在Activity中我们都推荐使用lifecycleScope来启动协程,lifecycleScope可以让协程具有与Activity一样的生命周期意识。

下面是lifecycleScope源码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
val LifecycleOwner.lifecycleScope: LifecycleCoroutineScope
get() = lifecycle.coroutineScope

val Lifecycle.coroutineScope: LifecycleCoroutineScope
get() {
while (true) {
val existing = mInternalScopeRef.get() as LifecycleCoroutineScopeImpl?
if (existing != null) {
return existing
}
val newScope = LifecycleCoroutineScopeImpl(
this,
SupervisorJob() + Dispatchers.Main.immediate
)
if (mInternalScopeRef.compareAndSet(null, newScope)) {
newScope.register()
return newScope
}
}
}

它创建了一个LifecycleCoroutineScopeImpl实例,它实现了CoroutineScope接口,同时传入SupervisorJob() + Dispatchers.Main作为它的CoroutineContext。

我们再来看它的register()方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
internal class LifecycleCoroutineScopeImpl(
override val lifecycle: Lifecycle,
override val coroutineContext: CoroutineContext
) : LifecycleCoroutineScope(), LifecycleEventObserver {
init {
// in case we are initialized on a non-main thread, make a best effort check before
// we return the scope. This is not sync but if developer is launching on a non-main
// dispatcher, they cannot be 100% sure anyways.
if (lifecycle.currentState == Lifecycle.State.DESTROYED) {
coroutineContext.cancel()
}
}

fun register() {
// TODO use Main.Immediate once it is graduated out of experimental.
launch(Dispatchers.Main) {
if (lifecycle.currentState >= Lifecycle.State.INITIALIZED) {
lifecycle.addObserver(this@LifecycleCoroutineScopeImpl)
} else {
coroutineContext.cancel()
}
}
}

override fun onStateChanged(source: LifecycleOwner, event: Lifecycle.Event) {
if (lifecycle.currentState <= Lifecycle.State.DESTROYED) {
lifecycle.removeObserver(this)
coroutineContext.cancel()
}
}

在register方法中通过经典的launch来创建一个协程,而launch使用到的CoroutineContext就是CoroutineSope中的CoroutineContext。然后在协程中结合Jetpack的Lifecycle特性来监听Activiyt的生命周期。

如果对Lifecycle的使用与特性还不是很了解的,推荐阅读这篇入门级文章Android Architecture Components Part3:Lifecycle

意思就是说在Activity销毁的时候会调用下面的方法取消协程的运行。

1
coroutineContext.cancel()

这里就使用到了CoroutineContext,经过上篇文章的分析我们很容易知道CoroutineContext自身是没有cancel方法的,所以这个cancel方法是CoroutineContext的扩展方法。

1
2
3
public fun CoroutineContext.cancel(): Unit {
this[Job]?.cancel()
}

所以真正的逻辑是从CoroutineContex集合中取出Key为Job的实例,这个对应的就是上面创建LifecycleCoroutineScopeImpl实例时传入的SupervisorJob,它是CoroutineContext的其中一个子类。

这时再来看lifecycleScope相关的一些方法

1
2
3
lifecycleScope.launchWhenCreated {  }
lifecycleScope.launchWhenStarted { }
lifecycleScope.launchWhenResumed { }

这些方法的内部逻辑就很明显了,也就是通过Lifecycle来追踪Activity的生命周期,从而约束协程运行的时机。

我们也可以不使用lifecycleScope,自己实现一个CoroutineScope,让它在Activity达到同样的效果。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
class MyActivity : AppCompatActivity(), CoroutineScope {
lateinit var job: Job
override val coroutineContext: CoroutineContext
get() = Dispatchers.Main + job
override fun onCreate(savedInstanceState: Bundle?) {
super.onCreate(savedInstanceState)
job = Job()
}
override fun onDestroy() {
super.onDestroy()
job.cancel() // Cancel job on activity destroy. After destroy all children jobs will be cancelled automatically
}
/*
* Note how coroutine builders are scoped: if activity is destroyed or any of the launched coroutines
* in this method throws an exception, then all nested coroutines are cancelled.
*/
fun loadDataFromUI() = launch { // <- extension on current activity, launched in the main thread
val ioData = async(Dispatchers.IO) { // <- extension on launch scope, launched in IO dispatcher
// blocking I/O operation
}
// do something else concurrently with I/O
val data = ioData.await() // wait for result of I/O
draw(data) // can draw in the main thread
}
}

上面的实现也能够保证当前Activiyt中的协程在Activity销毁的时候终止协程的运行。

到这里CoroutineScope的作用就呼之欲出了,它就是用来约束协程的边界,能够很好的提供对应的协程取消功能,保证协程的运行范围。

当然这又引申出另外一个话题

Job是什么?

Job

基本上每启动一个协程就会产生对应的Job,例如

1
2
lifecycleScope.launch {
}

launch返回的就是一个Job,它可以用来管理协程,一个Job中可以关联多个子Job,同时它也提供了通过外部传入parent的实现

1
public fun Job(parent: Job? = null): Job = JobImpl(parent)

这个很好理解,当传入parent时,此时的Job将会作为parent的子Job。

既然Job是来管理协程的,那么它提供了六种状态来表示协程的运行状态。

  1. New: 创建
  2. Active: 运行
  3. Completing: 已经完成等待自身的子协程
  4. Completed: 完成
  5. Cancelling: 正在进行取消或者失败
  6. Cancelled: 取消或失败

这六种状态Job对外暴露了三种状态,它们随时可以通过Job来获取

1
2
3
public val isActive: Boolean
public val isCompleted: Boolean
public val isCancelled: Boolean

所以如果你需要自己来手动管理协程,可以通过下面的方式来判断当前协程是否在运行。

1
2
3
while (job.isActive) {
// 协程运行中
}

一般来说,协程创建的时候就处在Active状态,但也有特例。

例如我们通过launch启动协程的时候可以传递一个start参数

1
2
3
4
5
6
7
public fun CoroutineScope.launch(
context: CoroutineContext = EmptyCoroutineContext,
start: CoroutineStart = CoroutineStart.DEFAULT,
block: suspend CoroutineScope.() -> Unit
): Job {
...
}

如果这个start传递的是CoroutineStart.LAZY,那么它将处于New状态。可以通过调用start或者join来唤起协程进入Active状态。

下面我们来看一张简图,就能很清晰的了解Job中的六个状态间的转化过程。

1
2
3
4
5
6
7
8
9
10
11
                                      wait children
+-----+ start +--------+ complete +-------------+ finish +-----------+
| New | -----> | Active | ---------> | Completing | -------> | Completed |
+-----+ +--------+ +-------------+ +-----------+
| cancel / fail |
| +----------------+
| |
V V
+------------+ finish +-----------+
| Cancelling | --------------------------------> | Cancelled |
+------------+ +-----------+

上面已经提及到一个Job可以有多个子Job,所以一个Job的完成都必须等待它内部所有的子Job完成;对应的cancel也是一样的。

默认情况下,如果内部的子Job发生异常,那么它对应的parent Job与它相关连的其它子Job都将取消运行。俗称连锁反应。

我们也可以改变这种默认机制,Kotlin提供了SupervisorJob来改变这种机制。这种情况还是很常见的,例如用协程请求两个接口,但并不想因为其中一个接口失败导致另外的接口也不请求,这时就可以使用SupervisorJob来改变协程的这种默认机制。

使用很简单,在我们创建CoroutineContext的时候加入SupervisorJob即可。例如在上面提到过的lifecycleScope,内部就使用到了SupervisorJob

1
2
3
4
val newScope = LifecycleCoroutineScopeImpl(
this,
SupervisorJob() + Dispatchers.Main
)

你也可以尝试运行下面的这个例子,然后将它的SupervisorJob替换成别的CoroutineContext再来看下效果。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
fun main() = runBlocking {
val supervisor = SupervisorJob()
with(CoroutineScope(coroutineContext + supervisor)) {
// 启动第一个子作业——这个示例将会忽略它的异常(不要在实践中这么做!)
val firstChild = launch(CoroutineExceptionHandler { _, _ -> }) {
println("The first child is failing")
throw AssertionError("The first child is cancelled")
}
// 启动第二个子作业
val secondChild = launch {
firstChild.join()
// 取消了第一个子作业且没有传播给第二个子作业
println("The first child is cancelled: ${firstChild.isCancelled}, but the second one is still active")
try {
delay(Long.MAX_VALUE)
} finally {
// 但是取消了监督的传播
println("The second child is cancelled because the supervisor was cancelled")
}
}
// 等待直到第一个子作业失败且执行完成
firstChild.join()
println("Cancelling the supervisor")
supervisor.cancel()
secondChild.join()
}
}

如果有些任务你并不想被手动取消,可以使用NonCancellable作为任务的CoroutineContext

如果需要Job获取协程的返回结果,可以通过Deferred来实现,它是Job的一个子类,所以也拥有Job所用功能。同时额外提供await方法来等待协程结果的返回。

Deferred可以通过CoroutineScope.async创建。

最后我们再来介绍下Job的几个方法,start与cancel就不多说了,分别是启动与取消。

invokeOnCompletion

这个方法是Job的回调通知,当Job执行完后会调用这个方法

1
2
3
public fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle

public typealias CompletionHandler = (cause: Throwable?) -> Unit

这个cause有三种情况分别为:

  1. is null: 协程正常执行完毕
  2. is CancellationException: 协程正常取消,并非异常导致的取消
  3. Otherwise: 协程发生异常

同时它的返回值DisposableHandle可以用来取消回调的监听。

join

1
public suspend fun join()

注意这是一个suspend函数,所以它只能在suspend或者coroutine中进行调用。

它的作用是暂停当前运行的协程任务,立刻执行自身Job的协程任务,直到自身执行完毕之后才恢复之前的协程任务继续执行。

本篇文章主要介绍了CoroutineScope的作用与Job的相关状态演化与运用。希望对学习协程的伙伴们能够有所帮助,敬请期待后续的协程分析。

项目

android_startup: 提供一种在应用启动时能够更加简单、高效的方式来初始化组件,优化启动速度。不仅支持Jetpack App Startup的全部功能,还提供额外的同步与异步等待、线程控制与多进程支持等功能。

AwesomeGithub: 基于Github的客户端,纯练习项目,支持组件化开发,支持账户密码与认证登陆。使用Kotlin语言进行开发,项目架构是基于JetPack&DataBinding的MVVM;项目中使用了Arouter、Retrofit、Coroutine、Glide、Dagger与Hilt等流行开源技术。

flutter_github: 基于Flutter的跨平台版本Github客户端,与AwesomeGithub相对应。

android-api-analysis: 结合详细的Demo来全面解析Android相关的知识点, 帮助读者能够更快的掌握与理解所阐述的要点。

daily_algorithm: 每日一算法,由浅入深,欢迎加入一起共勉。

支持一下
赞赏是一门艺术