Kotlin协程上下文与调度器源码级分析(35)

KOTLIN协程上下文与调度器源码级分析

一、Kotlin协程上下文概述

Kotlin协程的设计中，上下文（CoroutineContext）与调度器（CoroutineDispatcher）是两个核心概念，它们共同决定了协程的执行环境与行为模式。协程上下文是一个不可变的集合，包含了协程的各种元素，如Job、CoroutineDispatcher、CoroutineName等；而调度器则是上下文的重要组成部分，负责决定协程在哪个线程上执行。

1.1 协程上下文的基本概念

协程上下文是一个接口，定义了协程执行时的环境信息。它是一个元素的集合，每个元素都实现了Element接口，并且有唯一的Key。这种设计使得协程上下文可以包含多种不同类型的元素，同时保证元素的唯一性。

从源码角度看，CoroutineContext接口定义了以下核心方法：

public interface CoroutineContext {// 通过Key获取上下文中的元素public operator fun <E : Element> get(key: Key<E>): E?// 合并两个上下文public operator fun plus(context: CoroutineContext): CoroutineContext// 移除指定Key的元素public fun minusKey(key: Key<*>): CoroutineContext// 上下文元素的接口public interface Element : CoroutineContext {public val key: Key<*>override operator fun <E : Element> get(key: Key<E>): E? =@Suppress("UNCHECKED_CAST")if (this.key == key) this as E else nulloverride fun plus(context: CoroutineContext): CoroutineContext =context.fold(this) { acc, element ->val removed = acc.minusKey(element.key)if (removed === EmptyCoroutineContext) element else CombinedContext(removed, element)}override fun minusKey(key: Key<*>): CoroutineContext =if (this.key == key) EmptyCoroutineContext else this}// 上下文元素的Key接口public interface Key<E : Element>
}

1.2 调度器在协程中的作用

调度器是协程上下文的重要组成部分，它实现了CoroutineDispatcher抽象类，负责决定协程在哪个线程或线程池上执行。调度器可以将协程调度到不同的线程环境中，如主线程、IO线程池或自定义线程池。

调度器的核心接口定义如下：

public abstract class CoroutineDispatcher : ContinuationInterceptor {// 判断当前调度器是否在当前线程上运行public open fun isDispatchNeeded(context: CoroutineContext): Boolean = true// 将协程任务分派到指定的线程执行public abstract fun dispatch(context: CoroutineContext, block: Runnable)// 在特定情况下，调度器可以选择直接在当前线程执行协程public open fun dispatchYield(context: CoroutineContext, block: Runnable): Unit = dispatch(context, block)// 关闭调度器，释放资源public open fun close() {throw UnsupportedOperationException("Close is not supported for $this")}// 表示调度器的调试字符串public open val name: String? get() = nulloverride val key: CoroutineContext.Key<*> get() = CoroutineDispatcheroverride fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> =DispatchedContinuation(this, continuation)
}

二、协程上下文的源码结构

2.1 核心接口与类的定义

协程上下文的核心接口是CoroutineContext，它定义了上下文的基本操作。除了前面提到的方法外，CoroutineContext还定义了一些重要的实现类和伴生对象。

2.1.1 EmptyCoroutineContext

EmptyCoroutineContext是一个单例对象，表示空的协程上下文。它实现了CoroutineContext接口，并提供了高效的空上下文操作实现。

public object EmptyCoroutineContext : CoroutineContext, Serializable {private const val serialVersionUID: Long = 0Loverride fun <E : Element> get(key: Key<E>): E? = nulloverride fun plus(context: CoroutineContext): CoroutineContext = contextoverride fun minusKey(key: Key<*>): CoroutineContext = thisoverride fun iterator(): Iterator<Element> = emptyList<Element>().iterator()override fun hashCode(): Int = 0override fun toString(): String = "EmptyCoroutineContext"private fun readResolve(): Any = EmptyCoroutineContext
}

2.1.2 CombinedContext

CombinedContext是一个表示多个上下文元素组合的类。它使用链表结构存储多个元素，每个节点包含一个元素和指向下一个节点的引用。

private class CombinedContext(private val left: CoroutineContext,private val element: Element
) : CoroutineContext, Serializable {override fun <E : Element> get(key: Key<E>): E? {var cur = thiswhile (true) {cur.element[key]?.let { return it }val next = cur.leftif (next is CombinedContext) {cur = next} else {return next[key]}}}// 其他方法实现...
}

2.2 上下文元素的类型

协程上下文可以包含多种类型的元素，每种元素都有特定的作用。常见的上下文元素类型包括：

2.2.1 Job

Job是协程的一个重要组成部分，它表示协程的生命周期。Job接口继承自CoroutineContext.Element，其Key为Job.Key。

2.2.2 CoroutineDispatcher

CoroutineDispatcher负责协程的调度，决定协程在哪个线程上执行。它也是CoroutineContext.Element的实现，Key为CoroutineDispatcher。

2.2.3 CoroutineName

CoroutineName用于为协程指定一个名称，主要用于调试目的。它同样实现了CoroutineContext.Element接口，Key为CoroutineName.Key。

2.2.4 ContinuationInterceptor

ContinuationInterceptor用于拦截协程的延续（continuation），可以在协程恢复执行时插入自定义逻辑。它的Key为ContinuationInterceptor.Key。

三、调度器的源码分析

3.1 调度器的基本接口

CoroutineDispatcher是所有调度器的基类，它定义了调度器的基本行为。除了前面提到的方法外，CoroutineDispatcher还提供了一些默认实现和工具方法。

例如，dispatch方法是调度器的核心方法，它负责将协程任务分派到指定的线程执行。不同的调度器实现会以不同的方式实现这个方法。

3.2 内置调度器实现

Kotlin协程库提供了多种内置调度器，每种调度器都适用于不同的场景。

3.2.1 Dispatchers.Default

Dispatchers.Default是用于CPU密集型任务的调度器。它使用共享的线程池，默认线程数为CPU核心数。

源码实现中，Dispatchers.Default是通过DefaultScheduler类实现的，它继承自ExecutorCoroutineDispatcher：

internal object DefaultScheduler : ExecutorCoroutineDispatcher() {private const val DEFAULT_SCHEDULER_NAME = "DefaultDispatcher"// 默认线程数为CPU核心数，但不小于2private val DEFAULT_SCHEDULER_MAX_THREADS = System.getProperty("kotlinx.coroutines.default.parallelism")?.toIntOrNull()?.coerceAtLeast(1) ?: (Runtime.getRuntime().availableProcessors().coerceAtLeast(2))private val scheduler: Scheduler = createScheduler()// 创建调度器的方法private fun createScheduler(): Scheduler {return SchedulerImpl(parallelism = DEFAULT_SCHEDULER_MAX_THREADS,name = DEFAULT_SCHEDULER_NAME,corePoolSize = DEFAULT_SCHEDULER_MAX_THREADS,maxPoolSize = DEFAULT_SCHEDULER_MAX_THREADS,keepAliveTime = 60000L,threadFactory = { idx -> createPlainThreadFactory("$DEFAULT_SCHEDULER_NAME-worker-$idx") })}override val executor: Executor get() = scheduleroverride fun dispatch(context: CoroutineContext, block: Runnable) {scheduler.execute(block)}override fun close() {// 默认调度器不能关闭throw UnsupportedOperationException("Dispatchers.Default cannot be closed")}
}

3.2.2 Dispatchers.IO

Dispatchers.IO是用于IO密集型任务的调度器。它使用一个弹性线程池，线程数会根据需要动态调整。

源码实现中，Dispatchers.IO也是通过DefaultScheduler类实现的，但配置参数不同：

internal object IoScheduler : ExecutorCoroutineDispatcher() {private const val IO_SCHEDULER_NAME = "IO dispatcher"// IO调度器的默认线程数private val IO_PARALLELISM = System.getProperty("kotlinx.coroutines.io.parallelism")?.toIntOrNull()?.coerceAtLeast(1) ?: 64private val scheduler: Scheduler = createScheduler()// 创建调度器的方法private fun createScheduler(): Scheduler {return SchedulerImpl(parallelism = IO_PARALLELISM,name = IO_SCHEDULER_NAME,corePoolSize = 64,maxPoolSize = Int.MAX_VALUE,keepAliveTime = 60000L,threadFactory = { idx -> createPlainThreadFactory("$IO_SCHEDULER_NAME-worker-$idx") },autoAdjustPoolSize = true)}override val executor: Executor get() = scheduleroverride fun dispatch(context: CoroutineContext, block: Runnable) {scheduler.execute(block)}override fun close() {// IO调度器不能关闭throw UnsupportedOperationException("Dispatchers.IO cannot be closed")}
}

3.2.3 Dispatchers.Main

Dispatchers.Main是用于在主线程执行协程的调度器。在不同的平台上，它有不同的实现。

在Android平台上，Dispatchers.Main通过MainLooper实现：

@JvmField
public val Main: MainCoroutineDispatcher = createMainDispatcher()private fun createMainDispatcher(): MainCoroutineDispatcher {return if (runBlocking { isMainThread() }) {// 如果当前线程是主线程，使用直接调度器DirectMainDispatcher} else {// 否则使用基于Handler的调度器HandlerContext(Looper.getMainLooper(), "Main")}
}

在JavaFX平台上，Dispatchers.Main通过JavaFX的Platform.runLater方法实现：

internal object JavaFxDispatcher : MainCoroutineDispatcher() {override fun dispatch(context: CoroutineContext, block: Runnable) {if (Platform.isFxApplicationThread()) {block.run()} else {Platform.runLater(block)}}override fun close() {throw UnsupportedOperationException("Dispatchers.Main cannot be closed")}override val immediate: MainCoroutineDispatcher get() = this
}

3.3 自定义调度器

除了使用内置调度器，开发者还可以自定义调度器。自定义调度器需要继承CoroutineDispatcher类，并实现其抽象方法。

以下是一个简单的自定义调度器示例：

class MyDispatcher(private val executor: Executor) : CoroutineDispatcher() {override fun dispatch(context: CoroutineContext, block: Runnable) {executor.execute(block)}override fun close() {if (executor is ExecutorService) {executor.shutdown()}}
}

四、上下文元素的交互与协作

4.1 上下文元素的组合

协程上下文的组合是通过plus操作符实现的。当两个上下文组合时，会创建一个新的CombinedContext对象，其中包含两个上下文中的所有元素。

源码中，plus操作符的实现如下：

public operator fun CoroutineContext.plus(context: CoroutineContext): CoroutineContext =if (context === EmptyCoroutineContext) this else // 优化：如果右边是空上下文，直接返回左边if (this === EmptyCoroutineContext) context else // 优化：如果左边是空上下文，直接返回右边context.fold(this) { acc, element -> // 否则递归合并两个上下文val removed = acc.minusKey(element.key)if (removed === EmptyCoroutineContext) element else CombinedContext(removed, element)}

4.2 上下文元素的查找与替换

在协程上下文中查找元素是通过get方法实现的。该方法会递归遍历上下文链表，查找与指定Key匹配的元素。

源码中，get方法的实现如下：

public interface CoroutineContext {public operator fun <E : Element> get(key: Key<E>): E?// CombinedContext中的get方法实现private class CombinedContext(private val left: CoroutineContext,private val element: Element) : CoroutineContext, Serializable {override fun <E : Element> get(key: Key<E>): E? {var cur = thiswhile (true) {cur.element[key]?.let { return it }val next = cur.leftif (next is CombinedContext) {cur = next} else {return next[key]}}}}
}

替换上下文中的元素是通过先移除旧元素，再添加新元素实现的。这也是通过plus操作符完成的，因为plus操作符会自动处理元素的替换。

4.3 上下文元素之间的协作

不同的上下文元素之间可以相互协作，共同影响协程的行为。例如，Job元素和CoroutineDispatcher元素可以协同工作，实现协程的取消和调度。

当协程被取消时，Job元素会通知所有子协程，并通过调度器安排取消操作的执行。调度器则负责将取消操作分派到合适的线程上执行。

五、协程调度的实现机制

5.1 调度器的工作流程

协程调度的基本流程如下：

当协程挂起时，当前的延续（continuation）会被保存
当协程需要恢复执行时，调度器会被调用
调度器根据自身实现，将协程任务分派到合适的线程
协程在目标线程上恢复执行

源码中，这个过程通过DispatchedContinuation类实现：

internal class DispatchedContinuation<in T>(@JvmField val dispatcher: CoroutineDispatcher,@JvmField val continuation: Continuation<T>
) : Continuation<T> by continuation, CoroutineStackFrame {// 当协程恢复执行时调用override fun resumeWith(result: Result<T>) {val context = continuation.contextval state = result.toState()if (dispatcher.isDispatchNeeded(context)) {// 如果需要调度，将任务分派到调度器指定的线程_state = stateresumeMode = MODE_ATOMIC_DEFAULTdispatcher.dispatch(context, this)} else {// 否则直接在当前线程执行executeUnconfined(state, MODE_ATOMIC_DEFAULT) {completion.resumeWithStackTrace(it)}}}// 实现Runnable接口，用于在调度器中执行override fun run() {val state = _state_state = null // 清除状态if (state is CompletedExceptionally) {// 处理异常情况continuation.resumeWithException(state.cause)} else {// 正常恢复执行continuation.resumeWith(Result.success(state))}}
}

5.2 线程池的管理与优化

Kotlin协程的调度器使用线程池来管理线程资源。不同的调度器使用不同类型的线程池，以适应不同的场景。

例如，Dispatchers.Default使用固定大小的线程池，线程数为CPU核心数，适合CPU密集型任务；而Dispatchers.IO使用弹性线程池，线程数可以动态调整，适合IO密集型任务。

线程池的管理和优化是调度器实现的重要部分。例如，SchedulerImpl类实现了一个高效的线程池，它使用工作窃取算法来平衡线程负载：

private class SchedulerImpl(override val parallelism: Int,name: String,corePoolSize: Int,maxPoolSize: Int,keepAliveTime: Long,threadFactory: (Int) -> Thread,autoAdjustPoolSize: Boolean = false
) : ExecutorService, Scheduler {// 工作队列数组private val workQueues = Array(parallelism) { WorkQueue(it) }// 线程池执行器private val executor = ThreadPoolExecutor(corePoolSize,maxPoolSize,keepAliveTime,TimeUnit.MILLISECONDS,SynchronousQueue<Runnable>(),threadFactory).apply {allowCoreThreadTimeOut(true)}// 执行任务的方法override fun execute(command: Runnable) {val workerIndex = workerIndex.get()val queue = workQueues[workerIndex]if (!queue.offer(command)) {// 如果队列已满，将任务提交给线程池executor.execute(command)}}// 工作窃取方法fun steal(): Runnable? {val currentIndex = workerIndex.get()// 尝试从其他队列窃取任务for (i in 0 until parallelism) {val index = (currentIndex + i + 1) % parallelismval queue = workQueues[index]val task = queue.poll()if (task != null) {return task}}return null}// 其他方法实现...
}

5.3 调度器的切换与优化

在协程执行过程中，可能会发生调度器的切换。例如，协程可能先在IO调度器上执行IO操作，然后切换到主线程更新UI。

调度器的切换会带来一定的开销，因此Kotlin协程库对调度器切换进行了优化。例如，通过withContext方法切换调度器时，会检查目标调度器是否与当前调度器相同，如果相同则不会进行实际的切换。

源码中，withContext方法的实现如下：

public suspend fun <T> withContext(context: CoroutineContext,block: suspend CoroutineScope.() -> T
): T {// 如果提供的上下文为空，直接执行blockif (context === EmptyCoroutineContext) {return block()}// 获取当前协程上下文val currentContext = coroutineContext// 计算需要合并的上下文val newContext = currentContext + context// 检查是否需要切换上下文val oldDispatcher = currentContext[ContinuationInterceptor]val newDispatcher = newContext[ContinuationInterceptor]// 如果调度器相同且不需要额外的Job管理，直接执行blockif (oldDispatcher == newDispatcher && newContext[Job] == currentContext[Job]) {return block()}// 否则，创建一个新的协程作用域并执行blockreturn suspendCoroutineUninterceptedOrReturn sc@ { uCont ->val coroutine = ScopeCoroutine(newContext, uCont)coroutine.startUndispatchedOrReturn(coroutine, block)}
}

六、协程上下文与调度器的应用场景

6.1 CPU密集型任务的调度

对于CPU密集型任务，应该使用Dispatchers.Default调度器。这个调度器使用固定大小的线程池，线程数为CPU核心数，能够充分利用CPU资源。

例如，以下代码展示了如何在Dispatchers.Default上执行CPU密集型任务：

suspend fun calculatePi(digits: Int): String = withContext(Dispatchers.Default) {// 执行CPU密集型计算// ...
}

6.2 IO密集型任务的调度

对于IO密集型任务，应该使用Dispatchers.IO调度器。这个调度器使用弹性线程池，能够处理大量并发的IO操作。

例如，以下代码展示了如何在Dispatchers.IO上执行文件读取操作：

suspend fun readFile(path: String): String = withContext(Dispatchers.IO) {File(path).readText()
}

6.3 主线程任务的调度

在Android和JavaFX等UI框架中，更新UI的操作必须在主线程执行。这时可以使用Dispatchers.Main调度器。

例如，以下代码展示了如何在Android中使用Dispatchers.Main更新UI：

suspend fun fetchDataAndUpdateUI() {val data = withContext(Dispatchers.IO) {// 从网络获取数据fetchData()}withContext(Dispatchers.Main) {// 更新UIupdateUI(data)}
}

6.4 自定义调度器的应用

在某些特殊场景下，内置调度器可能无法满足需求，这时可以使用自定义调度器。

例如，以下代码展示了如何创建一个单线程调度器：

val singleThreadDispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()suspend fun runOnSingleThread() = withContext(singleThreadDispatcher) {// 在单线程中执行任务
}

七、协程上下文与调度器的性能考量

7.1 调度器切换的开销

调度器切换会带来一定的开销，包括线程上下文切换和任务排队的时间。因此，应该尽量减少不必要的调度器切换。

例如，在一个协程中多次调用withContext切换调度器会降低性能。如果可能，应该将相关操作放在同一个调度器中执行。

7.2 线程池大小的选择

线程池大小的选择对性能有重要影响。对于CPU密集型任务，线程池大小应该接近CPU核心数；对于IO密集型任务，线程池大小可以适当增大。

Kotlin协程的内置调度器已经针对不同场景进行了优化，但在使用自定义调度器时，需要根据具体场景选择合适的线程池大小。

7.3 避免线程饥饿

线程饥饿是指某些任务长时间无法获得执行所需的线程资源。为了避免线程饥饿，应该合理分配线程资源，避免在同一个调度器上执行过多长时间运行的任务。

例如，如果在Dispatchers.Default上执行大量长时间运行的任务，可能会导致其他CPU密集型任务无法获得足够的线程资源。

八、协程上下文与调度器的高级特性

8.1 非受限调度器（Unconfined）

Kotlin协程提供了一种特殊的调度器Dispatchers.Unconfined，它不会将协程调度到特定的线程，而是在当前线程上立即执行，直到第一个挂起点。

这种调度器适用于那些不需要特定线程环境的初始化操作，但在实际应用中应该谨慎使用，因为它可能导致协程在不同的线程上执行，增加调试难度。

源码中，Dispatchers.Unconfined的实现如下：

public object Unconfined : CoroutineDispatcher() {override fun isDispatchNeeded(context: CoroutineContext): Boolean = falseoverride fun dispatch(context: CoroutineContext, block: Runnable) {throw UnsupportedOperationException("Dispatchers.Unconfined.dispatch should not be called")}override fun dispatchYield(context: CoroutineContext, block: Runnable): Unit =throw UnsupportedOperationException("Dispatchers.Unconfined.dispatchYield should not be called")override fun toString(): String = "Dispatchers.Unconfined"override fun interceptContinuation(continuation: Continuation<*>): Continuation<*> =UnconfinedContinuation(continuation.context, continuation)
}

8.2 协程名称与调试

通过CoroutineName上下文元素，可以为协程指定一个名称，这在调试时非常有用。例如：

launch(CoroutineName("myCoroutine")) {// 协程代码
}

在调试时，协程名称会出现在线程名称和堆栈跟踪中，帮助开发者更容易地识别和调试问题。

8.3 协程拦截器

ContinuationInterceptor是一种特殊的上下文元素，它可以拦截协程的延续（continuation），在协程恢复执行时插入自定义逻辑。

例如，以下代码展示了如何实现一个简单的协程拦截器：

object LoggingInterceptor : ContinuationInterceptor {override val key: CoroutineContext.Key<*> = ContinuationInterceptoroverride fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> =LoggingContinuation(continuation)class LoggingContinuation<T>(private val continuation: Continuation<T>) : Continuation<T> by continuation {override fun resumeWith(result: Result<T>) {println("Before resume: $result")continuation.resumeWith(result)println("After resume")}}
}