协程作用域CoroutineScope
在 Android 环境中，通常每个界面（Activity、Fragment 等）启动的 Coroutine 只在该界面有意义，如果用户在等待 Coroutine 执行的时候退出了这个界面，则再继续执行这个 Coroutine 可能是没必要的。另外 Coroutine 也需要在适当的 context 中执行，否则会出现错误，比如在非 UI 线程去访问 View。 所以 Coroutine 在设计的时候，要求在一个范围（Scope）内执行，这样当这个 Scope 取消的时候，里面所有的子 Coroutine 也自动取消。所以要使用 Coroutine 必须要先创建一个对应的 CoroutineScope。
CoroutineScope 接口
CoroutineScope 是一个接口，要是查看这个接口的源代码的话就发现这个接口里面只定义了一个属性 CoroutineContext：
public interface CoroutineScope {
// Scope 的 Context
public val coroutineContext: CoroutineContext
}
所以 CoroutineScope 只是定义了一个新 Coroutine 的执行 Scope。每个协程coroutine builder(launch 、async等) 都是 CoroutineScope 的扩展方法，并且自动的继承了当前 Scope 的 coroutineContext 和取消操作。
public fun CoroutineScope.launch(
context: CoroutineContext = EmptyCoroutineContext,
start: CoroutineStart = CoroutineStart.DEFAULT,
block: suspend CoroutineScope.() -> Unit
): Job {
......
}
每个 coroutine builder 和 scope 方法（withContext、coroutineScope 等）都使用自己的 Scope 和 自己管理的 Job 来运行提供给这些函数的代码块。并且也会等待该代码块中所有子 Coroutine 执行，当所有子 Coroutine 执行完毕并且返回的时候， 该代码块才执行完毕，这种行为被称之为 “structured concurrency”。
全局作用域GlobalScope
GlobalScope 是 CoroutineScope 的一个单例实现，其代码也是非常简单的:
public object GlobalScope : CoroutineScope {
/**
* Returns [EmptyCoroutineContext].
*/
override val coroutineContext: CoroutineContext
get() = EmptyCoroutineContext
}
用法
GlobalScope.launch(Dispatchers.Main) {
delay(7000)
val result = "content"
//主线程里更新 UI
text.text = result
}
该实例所用的 CoroutineContext 是一个 EmptyCoroutineContext 实例（这也是一个单例 object 对象）。由于 GlobalScope 对象没有和应用生命周期组件相关联，需要自己管理 GlobalScope 所创建的 Coroutine，所以一般而言我们不直接使用 GlobalScope 来创建 Coroutine。
与生命周期绑定的作用域
一般而言，在应用中具有生命周期的组件应该实现 CoroutineScope 接口，并负责该组件内 Coroutine 的创建和管理。例如对于 Android 应用来说，可以在 Activity 中实现 CoroutineScope 接口， 例如：
class ScopedActivity : Activity(), CoroutineScope {
lateinit var job: Job
// CoroutineScope 的实现
override val coroutineContext: CoroutineContext
get() = Dispatchers.Main + job
override fun onCreate(savedInstanceState: Bundle?) {
super.onCreate(savedInstanceState)
job = Job()
/*
* 注意 coroutine builder 的 scope， 如果 activity 被销毁了或者该函数内创建的 Coroutine
* 抛出异常了，则所有子 Coroutines 都会被自动取消。不需要手工去取消。
*/
launch { // <- 自动继承当前 activity 的 scope context，所以在 UI 线程执行
val ioData = async(Dispatchers.IO) { // <- launch scope 的扩展函数，指定了 IO dispatcher，所以在 IO 线程运行
// 在这里执行阻塞的 I/O 耗时操作
}
// 和上面的并非 I/O 同时执行的其他操作
val data = ioData.await() // 等待阻塞 I/O 操作的返回结果
draw(data) // 在 UI 线程显示执行的结果
}
}
override fun onDestroy() {
super.onDestroy()
// 当 Activity 销毁的时候取消该 Scope 管理的 job。
// 这样在该 Scope 内创建的子 Coroutine 都会被自动的取消。
job.cancel()
}
}
由于所有的 Coroutine 都需要一个 CoroutineScope，所以为了方便创建 Coroutine，在 CoroutineScope 上有很多扩展函数，比如 launch、async、actor、cancel 等。
MainScope
在 Android 中会经常需要实现这个 CoroutineScope，所以为了方便开发者使用， 标准库中定义了一个 MainScope() 函数，该函数定义了一个使用 SupervisorJob 和 Dispatchers.Main 为 Scope context 的实现。所以上面的代码可以简化为：
class ScopedActivity : Activity(),
CoroutineScope by MainScope(){ // 使用 by 指定代理实现
override fun onDestroy() {
super.onDestroy()
cancel() // 调用 CoroutineScope 的 cancel 函数
}
在mvvm模式使用作用域
class ViewModelOne : ViewModel() {
private val viewModelJob = SupervisorJob()
private val uiScope = CoroutineScope(Dispatchers.Main + viewModelJob)
val mMessage: MutableLiveData<String> = MutableLiveData()
fun getMessage(message: String) {
uiScope.launch {
val deferred = async(Dispatchers.IO) {
delay(2000)
"post $message"
}
mMessage.value = deferred.await()
}
}
override fun onCleared() {
super.onCleared()
viewModelJob.cancel()
}
}
ViewModelScope 方式
AndroidX Lifecycle v2.1.0 在 ViewModel 中引入 viewModelScope，当 ViewModel 被销毁时它会自动取消协程任务，这个特性真的好用。viewModelScope 管理协程的方式与我们在 ViewModel 引入协程的方式一样，代码实现如下：
class MyViewModel : ViewModel() {
fun launchDataLoad() {
viewModelScope.launch {
sortList()
// Modify UI
}
}
suspend fun sortList() = withContext(Dispatchers.Default) {
// Heavy work
}
}
协程上下文CoroutineContext
CoroutineContext是一个接口，我们常用到的Job, Dispatchers都是实现了该接口的类，此外还包括 CoroutineName 和CoroutineId等类。
@SinceKotlin("1.3")
public interface CoroutineContext {
public operator fun <E : Element> get(key: Key<E>): E?
public fun <R> fold(initial: R, operation: (R, Element) -> R): R
public operator fun plus(context: CoroutineContext): CoroutineContext = ...
public fun minusKey(key: Key<*>): CoroutineContext
public interface Key<E : Element>
public interface Element : CoroutineContext {
public val key: Key<*>
...
}
}
类似于一个以 Key 为索引的 List：
CoroutineContext 作为一个集合，它的元素就是源码中看到的 Element，每一个 Element 都有一个 key，因此它可以作为元素出现，同时它也是 CoroutineContext 的子接口，因此也可以作为集合出现。
我们看下实现了上下文接口的Job类
public interface Job : CoroutineContext.Element {
/**
* Key for [Job] instance in the coroutine context.
*/
public companion object Key : CoroutineContext.Key<Job> {
init {
/*
* Here we make sure that CoroutineExceptionHandler is always initialized in advance, so
* that if a coroutine fails due to StackOverflowError we don't fail to report this error
* trying to initialize CoroutineExceptionHandler
*/
CoroutineExceptionHandler
}
}
}
可以看到Job实现关系是Job<=Element<=CoroutineContext
public interface Element : CoroutineContext {
/**
* A key of this coroutine context element.
*/
public val key: Key<*>
public override operator fun <E : Element> get(key: Key<E>): E? =
@Suppress("UNCHECKED_CAST")
if (this.key == key) this as E else null
public override fun <R> fold(initial: R, operation: (R, Element) -> R): R =
operation(initial, this)
public override fun minusKey(key: Key<*>): CoroutineContext =
if (this.key == key) EmptyCoroutineContext else this
}
协程生成器 Coroutine builders
我们知道生成协程的方式有很多种，比如 launch、async、runBlocking等，他们都是
CoroutineScope的扩展方法，且都会创建一个新的协程
下面我们拿launch开启一个协程的例子来了解下如何设置上下文
public fun CoroutineScope.launch(
context: CoroutineContext = EmptyCoroutineContext,
start: CoroutineStart = CoroutineStart.DEFAULT,
block: suspend CoroutineScope.() -> Unit
): Job {
val newContext = newCoroutineContext(context)
...
}
我们为上下文添加调度器与上下文名字
launch(context = Dispatchers.Main.plus(CoroutineName("jason"))) {
val job = coroutineContext.get(Job)
val dispatcher = coroutineContext.get(ContinuationInterceptor)
val name = coroutineContext.get(CoroutineName)
println("job:$job")
println("dispatcher:$dispatcher")
println("name:$name")
}
打印输出结果
job:StandaloneCoroutine{Active}@ad739cb
dispatcher:Main
name:CoroutineName(jason)
上下文切换器withContext
与 launch、async、runBlocking 等不同，withContext 不会创建新的协程。
withContext 允许切换协程上下文，使用时必须传递一个 CoroutineContext
public suspend fun <T> withContext(
context: CoroutineContext,
block: suspend CoroutineScope.() -> T
): T = suspendCoroutineUninterceptedOrReturn sc@ { uCont ->
...
}
看下例子
launch(context = Dispatchers.IO.plus(CoroutineName("c1"))) {
val dispatcher = coroutineContext[ContinuationInterceptor]
val name = coroutineContext[CoroutineName]
println("scope:$this")
println("dispatcher:$dispatcher")
println("name:$name")
withContext(Dispatchers.Main.plus(CoroutineName("c2"))){
val dispatcher2 = coroutineContext[ContinuationInterceptor]
val name2 = coroutineContext[CoroutineName]
println("scope2:$this")
println("dispatcher2:$dispatcher2")
println("name2:$name2")
}
}
日志输出结果
scope:StandaloneCoroutine{Active}@60d84a6
dispatcher:LimitingDispatcher@903ee7[dispatcher = DefaultDispatcher]
name:CoroutineName(c1)
scope2:DispatchedCoroutine{Active}@e3f5a94
dispatcher2:Main
name2:CoroutineName(c2)
续体拦截器ContinuationInterceptor
拦截器也是上下文的一种，它实现了上下文接口
我们可以用拦截器打日志等。调度器就是基于拦截器实现的，换句话说调度器就是拦截器的一种。
public interface ContinuationInterceptor : CoroutineContext.Element {
companion object Key : CoroutineContext.Key<ContinuationInterceptor>
public fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T>
...
}
下面我们自己定义一个拦截器放到我们的协程上下文中，看看会发生什么。
class MyContinuationInterceptor: ContinuationInterceptor {
override val key = ContinuationInterceptor
override fun <T> interceptContinuation(continuation: Continuation<T>) = MyContinuation(continuation)
}
class MyContinuation<T>(val continuation: Continuation<T>): Continuation<T> {
override val context = continuation.context
override fun resumeWith(result: Result<T>) {
Log.d("jason","result=$result" )
continuation.resumeWith(result)
}
}
launch {
launch(MyContinuationInterceptor()) {
log(1)
val deferred = async {
log(2)
delay(1000)
log(3)
"我是返回值"
}
log(4)
val result = deferred.await()
log("5. $result")
}.join()
log(6)
}
我们通过 launch 启动了一个协程，为它指定了我们自己的拦截器作为上下文，紧接着在其中用 async 启动了一个协程，async 与 launch 从功能上是同等类型的函数，它们都被称作协程的 Builder 函数，不同之处在于 async 启动的 Job 也就是实际上的 Deferred 可以有返回结果，可以通过 await 方法获取。
输出日志
23:02:58.595 5241-5241/com.example.mytest D/jason: result=Success(kotlin.Unit) //1
23:02:58.596 5241-5241/com.example.mytest D/jason: 1
23:02:58.598 5241-5241/com.example.mytest D/jason: result=Success(kotlin.Unit) //2
23:02:58.598 5241-5241/com.example.mytest D/jason: 2
23:02:58.602 5241-5241/com.example.mytest D/jason: 4
23:02:59.602 5241-5273/com.example.mytest D/jason: result=Success(kotlin.Unit) //3
23:02:59.602 5241-5273/com.example.mytest D/jason: 3
23:02:59.603 5241-5273/com.example.mytest D/jason: result=Success(我是返回值) //4
23:02:59.604 5241-5273/com.example.mytest D/jason: 5. 我是返回值
23:02:59.605 5241-5241/com.example.mytest D/jason: 6
首先，这段代码中一共有4次调度机会，所有协程启动的时候，都会有一次 Continuation.resumeWith 的操作，这一次操作对于调度器来说就是一次调度的机会，我们的协程有机会调度到其他线程的关键之处就在于此。 ①、② 两处都是这种情况。
其次，delay 是挂起点，1000ms 之后需要继续调度执行该协程，因此就有了 ③ 处的日志。
最后，④ 处的日志就很容易理解了，正是我们的返回结果。
如果我们在拦截器当中自己处理了线程切换，那么就实现了自己的一个简单的调度器，大家有兴趣可以自己去尝试。
协程任务执行环境-Dispatcher(调度器)
public abstract class CoroutineDispatcher :
AbstractCoroutineContextElement(ContinuationInterceptor), ContinuationInterceptor {
...
public abstract fun dispatch(context: CoroutineContext, block: Runnable)
...
}
它本身是协程上下文的子类，同时实现了拦截器的接口， dispatch 方法会在拦截器的方法 interceptContinuation 中调用，进而实现协程的调度。所以如果我们想要实现自己的调度器，继承这个类就可以了，不过通常我们都用现成的
现成的调度器有：
| 调度器名称 | 使用线程 |
|---|---|
| Main | UI线程 |
| IO | 线程池 |
| Default | 线程池 |
| Unconfined | 直接执行 |
也能使用这种方式简单的创建一个自定义的协程调度器
val myDispatcher= Executors.newSingleThreadExecutor{ r -> Thread(r, "MyThread") }.asCoroutineDispatcher()
这里我们主要看一下我们最常使用的Dispatcher.Main和Dispatcher.IO两个派发器。
image.png
Dispatcher.Main
Dispatcher.Main没有默认实现，依赖于各个平台的实现，如果没有引入android依赖包，则会抛异常提示，那么kotlin是怎么支持这种动态的类呢？
- 首先kotlin提供了一个工厂类接口，用来创建MainDispatcher
public interface MainDispatcherFactory {
fun createDispatcher(allFactories: List<MainDispatcherFactory>): MainCoroutineDispatcher
}
- 然后再看反编译的源码
public final <S> List<S> loadProviders$kotlinx_coroutines_core(@NotNull Class<S> paramClass, @NotNull ClassLoader paramClassLoader) {
//从apk的META-INF/services/文件夹下那类名
StringBuilder stringBuilder = new StringBuilder();
stringBuilder.append("META-INF/services/");
stringBuilder.append(paramClass.getName());
Enumeration enumeration = paramClassLoader.getResources(stringBuilder.toString());
ArrayList arrayList = Collections.list(enumeration);
Iterable iterable = (Iterable)arrayList;
Collection collection = (Collection)new ArrayList();
for (URL uRL : iterable) {
FastServiceLoader fastServiceLoader = INSTANCE;
Intrinsics.checkExpressionValueIsNotNull(uRL, "it");
CollectionsKt.addAll(collection, (Iterable)fastServiceLoader.parse(uRL));
}
collection = CollectionsKt.toSet((Iterable)collection);
iterable = (Iterable)collection;
collection = (Collection)new ArrayList(CollectionsKt.collectionSizeOrDefault(iterable, 10));
//将类名解析为实例对象
for (String str : iterable)
collection.add(INSTANCE.getProviderInstance(str, paramClassLoader, paramClass));
return (List)collection;
}
MainDispatcher的factory会从apk的META-INF/services/文件夹下获取。
-
再看编译生成的apk文件的该文件夹内容
image.png
所以android的依赖包是通过向该文件注册类名实现的注册类，并且factory类为AndroidDispatcherFactory。
- 最后我们再来看下AndroidDispatcherFactory类
internal class AndroidDispatcherFactory : MainDispatcherFactory {
override fun createDispatcher(allFactories: List<MainDispatcherFactory>) = HandlerContext(Looper.getMainLooper().asHandler(async = true), "Main")
}
internal class HandlerContext private constructor(
private val handler: Handler,
private val name: String?,
private val invokeImmediately: Boolean
) : HandlerDispatcher(), Delay {
public constructor(
handler: Handler,
name: String? = null
) : this(handler, name, false)
//android中需要向主looper进行提交调度
override fun isDispatchNeeded(context: CoroutineContext): Boolean {
return !invokeImmediately || Looper.myLooper() != handler.looper
}
//通过持有主线程looper的handler进行调度
override fun dispatch(context: CoroutineContext, block: Runnable) {
handler.post(block)
}
...
}
很清楚，就是用持有主线程looper的handler进行任务的调度，确保任务会在主线程执行。
Dispatcher.IO
internal object DefaultScheduler : ExperimentalCoroutineDispatcher() {
val IO = blocking(systemProp(IO_PARALLELISM_PROPERTY_NAME, 64.coerceAtLeast(AVAILABLE_PROCESSORS)))
public fun blocking(parallelism: Int = BLOCKING_DEFAULT_PARALLELISM): CoroutineDispatcher {
return LimitingDispatcher(this, parallelism, TaskMode.PROBABLY_BLOCKING)
}
}
Dispatcher.IO是一个LimitingDispatcher实例，他可以控制同时并发任务数，默认为64个，即最多有64个任务同时在运行。
private class LimitingDispatcher(
val dispatcher: ExperimentalCoroutineDispatcher,
val parallelism: Int,
override val taskMode: TaskMode
) : ExecutorCoroutineDispatcher()
而LimitingDispatcher内部真正调度任务的dispatcher是一个ExperimentalCoroutineDispatcher对象。
open class ExperimentalCoroutineDispatcher(
private val corePoolSize: Int,
private val maxPoolSize: Int,
private val idleWorkerKeepAliveNs: Long,
private val schedulerName: String = "CoroutineScheduler"
) : ExecutorCoroutineDispatcher() {
constructor(
corePoolSize: Int = CORE_POOL_SIZE,
maxPoolSize: Int = MAX_POOL_SIZE,
schedulerName: String = DEFAULT_SCHEDULER_NAME
) : this(corePoolSize, maxPoolSize, IDLE_WORKER_KEEP_ALIVE_NS, schedulerName)
private var coroutineScheduler = createScheduler()
override fun dispatch(context: CoroutineContext, block: Runnable): Unit =
try {
coroutineScheduler.dispatch(block)
} catch (e: RejectedExecutionException) {
DefaultExecutor.dispatch(context, block)
}
private fun createScheduler() = CoroutineScheduler(corePoolSize, maxPoolSize, idleWorkerKeepAliveNs, schedulerName)
}
我们看到，该dispatcher里面的真正的线程池，是CoroutineScheduler对象，而核心线程数和最大线程数，取决于可用CPU的数量。
internal val CORE_POOL_SIZE = systemProp(
"kotlinx.coroutines.scheduler.core.pool.size",
AVAILABLE_PROCESSORS.coerceAtLeast(2), // !!! at least two here
minValue = CoroutineScheduler.MIN_SUPPORTED_POOL_SIZE
)
internal val AVAILABLE_PROCESSORS = Runtime.getRuntime().availableProcessors()
协程调度器-CoroutineScheduler
这里我们挑几个小细节看一下CoroutineScheduler是如何来优化对线程的使用的。
i. 尽量使用当前线程
private fun submitToLocalQueue(task: Task, fair: Boolean): Int {
val worker = currentWorker() ?: return NOT_ADDED
...
worker.localQueue.add(task, globalQueue)
...
}
private fun currentWorker(): Worker? = (Thread.currentThread() as? Worker)?.takeIf { it.scheduler == this }
如果当前线程是Dispatcher.IO开启的工作线程，那么任务优先交由该线程的任务队列，等待处理。
ii. 双重队列
fun dispatch(block: Runnable, taskContext: TaskContext = NonBlockingContext, fair: Boolean = false) {
...
when (submitToLocalQueue(task, fair)) {
ADDED -> return
NOT_ADDED -> {
//本地队列已满，放入全局队列，所有线程可取
globalQueue.addLast(task)
}
else -> requestCpuWorker() // ask for help
}
}
如果工作线程本地队列无限大，一味的放入本地队列的话，可能会造成单一线程工作，效率极低，于是每个工作线程有固定大小的queue，满了之后，会放到全局queue中，等待任意空闲工作线程执行。
iii.抢占其他线程的任务
//工作线程Worker类
override fun run() {
while (!isTerminated && state != WorkerState.TERMINATED) {
val task = findTask()
...
}
...
}
private fun findTaskWithCpuPermit(): Task? {
...
//从本地queue获取任务
localQueue.poll()?.let { return it }
//从全局queue获取任务
if (!globalFirst) globalQueue.removeFirstOrNull()?.let { return it }
//抢占其他线程任务
return trySteal()
}
private fun trySteal(): Task? {
...
//随机一个工作线程
if (stealIndex == 0) stealIndex = nextInt(created)
...
val worker = workers[stealIndex]
if (worker !== null && worker !== this) {
//将其queue里的任务放到自己queue中
if (localQueue.trySteal(worker.localQueue, globalQueue)) {
return localQueue.poll()
}
}
return null
}
如果一个工作线程的本地queue和全局queue都没有任务了，但是其他线程的queue还有任务，此时让其空闲，一是没有充分利用线程提升工作效率，二是线程的空闲状态切换需要开销，所以此时会尝试从任一工作线程的queue中取出任务，放入自己的queue中执行。
以上三点的相互配合，可以充分利用线程资源，避免过多线程的使用及开销，也保证了多任务时的工作效率。
协程执行过程源码追踪分析
我们以一个请求数据后在主线程更新界面的代码来进行分析
fun setUpUI(){
GlobalScope.launch(Main) {
val dataDeferred = requestDataAsync()
doSomethingElse()
val data = dataDeferred.await()
processData(data)
}
Thread.sleep(1000)
doSomethingElse2()
}
fun requestDataAsync():Deferred<String>{
// 启动一个异步协程去执行耗时任务
return GlobalScope.async {
requestData()
}
}
fun doSomethingElse2(){
println("doSomethingElse2")
}
编译后生成伪代码
final void setUpUI() {
BuildersKt__Builders_commonKt.launch$default(GlobalScope.INSTANCE,
Dispatchers.getMain(),
null,
// 传入的是一个 KotlinTest$setUpUI.KotlinTest$setUpUI$1 对象
(Function2)new KotlinTest$setUpUI.KotlinTest$setUpUI$1(this, (Continuation)null), 2, null);
this.doSomethingElse2();
}
final class setUpUI$1 extends SuspendLambda implements Function2{
public final Object invokeSuspend(Object result) {
switch (this.label) {
case 0:
doSomethingElse()
// 新建并启动 async 协程
Deferred async$default = BuildersKt.async$default(coroutineScope, (CoroutineContext) Dispatchers.getDefault(), null, (Function2) new 1(null), 2, null);
this.label = 1;
// 如果 async 协程还没完成为挂起状态 则直接返回，等待下次唤醒重入
if (async$default.await(this) == coroutine_suspended) {
return coroutine_suspended;
}
break;
case 1:
val data = result;
processData(data)
break;
}
}
}
可以看到传入到 launch 函数第四个参数位置的是一个编译后生成的 SuspendLambda 类实例setUpUI$1，SuspendLambda 本质上是一个续体 Continuation，而 Continuation 是一个有着恢复操作的接口
/**
* 在一个挂起点之后可以返回类型T值的续集continuation的接口
* Interface representing a continuation after a suspension point that returns value of type `T`.
*/
@SinceKotlin("1.3")
public interface Continuation<in T> {
/**
* Context of the coroutine that corresponds to this continuation.
*/
// todo: shall we provide default impl with EmptyCoroutineContext?
public val context: CoroutineContext
/**
* Resumes the execution of the corresponding coroutine passing successful or failed [result] as the
* return value of the last suspension point.
*/
public fun resumeWith(result: Result<T>)
}
SuspendLambda 继承结构如下
SuspendLambda > ContinuationImpl > BaseContinuationImpl > Continuation
每一层封装对应添加了不同的功能，我们先忽略掉这些功能细节，着眼于我们的主线，继续跟进launch 函数执行过程，由于第二个参数是默认值，所以创建的是 StandaloneCoroutine， 调用链如下：
coroutine.start(start, coroutine, block)
-> CoroutineStart.start(block, receiver, this)
-> CoroutineStart.invoke(block: suspend () -> T, completion: Continuation<T>)
-> block.startCoroutineCancellable(completion)
-> createCoroutineUnintercepted(completion).intercepted().resumeCancellable(Unit)
我们看最后创建了一个协程，并链式调用 intercepted、resumeCancellable 方法，利用协程上下文中的续体拦截器 ContinuationInterceptor 对协程的执行进行拦截，intercepted 实际上调用的是 ContinuationImpl 的 intercepted 方法
internal abstract class ContinuationImpl(
completion: Continuation<Any?>?,
private val _context: CoroutineContext?
) : BaseContinuationImpl(completion) {
...
public fun intercepted(): Continuation<Any?> =
intercepted
?: (context[ContinuationInterceptor]?.interceptContinuation(this) ?: this)
.also { intercepted = it }
...
}
context[ContinuationInterceptor]?.interceptContinuation调用的是 CoroutineDispatcher 的 interceptContinuation 方法
public final <T> Continuation<T> interceptContinuation(@NotNull final Continuation<? super T> continuation) {
Intrinsics.checkParameterIsNotNull(continuation, "continuation");
return new DispatchedContinuation<T>(this, continuation);
}
最终创建了一个 DispatchedContinuation 可分发的协程实例，我们继续看resumeCancellable 方法
internal fun <T> Continuation<T>.resumeCancellable(value: T) = when (this) {
// 判断是否是DispatchedContinuation 根据我们前面的代码追踪 这里是DispatchedContinuation
is DispatchedContinuation -> resumeCancellable(value)
else -> resume(value)
}
inline fun resumeCancellable(value: T) {
// 判断是否需要线程调度
if (dispatcher.isDispatchNeeded(context)) {
_state = value
resumeMode = MODE_CANCELLABLE
dispatcher.dispatch(context, this)
} else {
UndispatchedEventLoop.execute(this, value, MODE_CANCELLABLE) {
if (!resumeCancelled()) {
resumeUndispatched(value)
}
}
}
}
最终走到 dispatcher.dispatch(context, this) 而这里的 dispatcher 就是通过工厂方法创建的 HandlerDispatcher ，dispatch() 函数第二个参数this是一个runnable这里为 DispatchedTask
HandlerDispatcher
/**
* Implements [CoroutineDispatcher] on top of an arbitrary Android [Handler].
*/
internal class HandlerContext private constructor(
private val handler: Handler,
private val name: String?,
private val invokeImmediately: Boolean
) : HandlerDispatcher(), Delay {
...
// 最终执行这里的 dispatch方法 而handler则是android中的 MainHandler
override fun dispatch(context: CoroutineContext, block: Runnable) {
handler.post(block)
}
...
}
这里借用 Android 的主线程消息队列来在主线程中执行 block Runnable 而这个 Runnable 即为 DispatchedTask
internal abstract class DispatchedTask<in T>(
@JvmField var resumeMode: Int
) : SchedulerTask() {
...
public final override fun run() {
...
withCoroutineContext(context, delegate.countOrElement) {
if (job != null && !job.isActive)
// 异常情况下
continuation.resumeWithException(job.getCancellationException())
else {
val exception = getExceptionalResult(state)
if (exception != null)
// 异常情况下
continuation.resumeWithStackTrace(exception)
else
// 正常情况下走到这一步
continuation.resume(getSuccessfulResult(state))
}
}
...
}
}
@InlineOnly public inline fun <T> Continuation<T>.resume(value: T): Unit =
resumeWith(Result.success(value))
internal abstract class BaseContinuationImpl(...) {
// 实现 Continuation 的 resumeWith，并且是 final 的，不可被重写
public final override fun resumeWith(result: Result<Any?>) {
...
val outcome = invokeSuspend(param)
...
}
// 由编译生成的协程相关类来实现，例如 setUpUI$1
protected abstract fun invokeSuspend(result: Result<Any?>): Any?
}
最终调用到 continuation.resumeWith() 而 resumeWith() 中会调用 invokeSuspend，即之前编译器生成的 SuspendLambda 中的 invokeSuspend 方法
final class setUpUI$1 extends SuspendLambda implements Function2{
public final Object invokeSuspend(Object result) {
switch (this.label) {
case 0:
doSomethingElse()
// 新建并启动 async 协程
Deferred async$default = BuildersKt.async$default(coroutineScope, (CoroutineContext) Dispatchers.getDefault(), null, (Function2) new 1(null), 2, null);
this.label = 1;
// 如果 async 协程还没完成为挂起状态 则直接返回，等待下次唤醒重入
if (async$default.await(this) == coroutine_suspended) {
return coroutine_suspended;
}
break;
case 1:
val data = result;
processData(data)
break;
}
}
}
这段代码是一个状态机机制，每一个挂起点都是一种状态，协程恢复只是跳转到下一个状态，挂起点将执行过程分割成多个片段，利用状态机的机制保证各个片段按顺序执行。
如果没有挂起点就只有一个初始状态，类似于callback回调，所以对应了之前我们分析的非阻塞的异步底层实现其实也是一种callback回调，只不过有多个挂起点时就会有多个callback回调，我们把多个callback回调封装成了一个状态机。
协程的挂起
从协程的调度过程我们知道，调度后会到编译器生成的 SuspendLambda 的 invokeSuspend 方法中的一个挂起方法，以例子中的await为例
if (async$default.await(this) == coroutine_suspended) {
//目前还在挂起中，则return等待挂起结束后的invokeSuspend
return coroutine_suspended;
}
async 也是一个协程，如果状态为挂起coroutine_suspended，则执行流直接 return 返回，如果已达到完成状态直接跳转下一个状态 case 1 最终走完整个协程代码块。
这里需要注意的是：
启动一个新的协程并不会挂起当前协程，只有当使用库函数 await、yield方法时才会将当前的协程挂起。
协程挂起并不会阻塞线程，线程在挂起点 return 后可以去执行其他的代码块。
协程的挂起过程很简单，代码块直接返回，当前状态保存在状态机 SuspendLambda 中，可以想象到协程恢复的时候也是调用 SuspendLambda 的 invokeSuspend 从而进入下一个状态继续执行的。
delay 的实现
public suspend fun delay(timeMillis: Long) {
if (timeMillis <= 0) return // don't delay
return suspendCancellableCoroutine sc@ { cont: CancellableContinuation<Unit> ->
cont.context.delay.scheduleResumeAfterDelay(timeMillis, cont)
}
}
/** Returns [Delay] implementation of the given context */
internal val CoroutineContext.delay: Delay get() = get(ContinuationInterceptor) as? Delay ?: DefaultDelay
internal actual val DefaultDelay: Delay = DefaultExecutor
delay 使用suspendCancellableCoroutine挂起协程，而协程恢复的一般情况下是关键在DefaultExecutor.scheduleResumeAfterDelay()，其中实现是schedule(DelayedResumeTask(timeMillis, continuation))，其中的关键逻辑是将 DelayedResumeTask 放到 DefaultExecutor 的队列最后，在延迟的时间到达就会执行 DelayedResumeTask，那么该 task 里面的实现是什么：
override fun run() {
// 直接在调用者线程恢复协程
with(cont) { resumeUndispatched(Unit) }
}
yield 的实现
yield()的作用是挂起当前协程，然后将协程分发到 Dispatcher 的队列，这样可以让该协程所在线程或线程池可以运行其他协程逻辑，然后在 Dispatcher 空闲的时候继续执行原来协程。简单的来说就是让出自己的执行权，给其他协程使用，当其他协程执行完成或也让出执行权时，一开始的协程可以恢复继续运行。
看下面的代码示例：
fun main(args: Array<String>) = runBlocking<Unit> {
launch {
repeat(3) {
println("job1 repeat $it times")
yield()
}
}
launch {
repeat(3) {
println("job2 repeat $it times")
yield()
}
}
}
通过yield()实现 job1 和 job2 两个协程交替运行，输出如下：
job1 repeat 0 times
job2 repeat 0 times
job1 repeat 1 times
job2 repeat 1 times
job1 repeat 2 times
job2 repeat 2 times
现在来看其实现：
public suspend fun yield(): Unit = suspendCoroutineUninterceptedOrReturn sc@ { uCont ->
val context = uCont.context
// 检测协程是否已经取消或者完成，如果是的话抛出 CancellationException
context.checkCompletion()
// 如果协程没有线程调度器，或者像 Dispatchers.Unconfined 一样没有进行调度，则直接返回
val cont = uCont.intercepted() as? DispatchedContinuation<Unit> ?: return@sc Unit
if (!cont.dispatcher.isDispatchNeeded(context)) return@sc Unit
// dispatchYield(Unit) 最终会调用到 dispatcher.dispatch(context, block) 将协程分发到调度器队列中，这样线程可以执行其他协程
cont.dispatchYield(Unit)
COROUTINE_SUSPENDED
}
所以注意到，yield()需要依赖协程的线程调度器，而调度器再次执行该协程时，会调用resume来恢复协程运行。
现在来看封装异步逻辑为挂起函数的关键是用suspendCoroutineUninterceptedOrReturn函数包装，然后在异步逻辑完成时调用resume手动恢复协程。
协程工作流程图
image.png
网友评论