协程作用域、上下文与调度

协程作用域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是怎么支持这种动态的类呢？

  1. 首先kotlin提供了一个工厂类接口，用来创建MainDispatcher
public interface MainDispatcherFactory {
    fun createDispatcher(allFactories: List<MainDispatcherFactory>): MainCoroutineDispatcher
}
  1. 然后再看反编译的源码
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/文件夹下获取。

  1. 再看编译生成的apk文件的该文件夹内容


    image.png

    所以android的依赖包是通过向该文件注册类名实现的注册类，并且factory类为AndroidDispatcherFactory。

  2. 最后我们再来看下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
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 193,968评论 5赞 459
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 81,682评论 2赞 371
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 141,254评论 0赞 319
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 52,074评论 1赞 263
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 60,964评论 4赞 355
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 46,055评论 1赞 272
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 36,484评论 3赞 381
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 35,170评论 0赞 253
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 39,433评论 1赞 290
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 34,512评论 2赞 308
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 36,296评论 1赞 325
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 32,184评论 3赞 312
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 37,545评论 3赞 298
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 28,880评论 0赞 17
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 30,150评论 1赞 250
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 41,437评论 2赞 341
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 40,630评论 2赞 335

推荐阅读更多精彩内容