Kotlin协程之Dispatchers原理
Kotlin协程不是什么空中阁楼，Kotlin源代码会被编译成class字节码文件，最终会运行到虚拟机中。所以从本质上讲，Kotlin和Java是类似的，都是可以编译产生class的语言，但最终还是会受到虚拟机的限制，它们的代码最终会在虚拟机上的某个线程上被执行。
之前我们分析了launch的原理，但当时我们没有去分析协程创建出来后是如何与线程产生关联的，怎么被分发到具体的线程上执行的，本篇文章就带大家分析一下。
前置知识
要想搞懂Dispatchers，我们先来看一下Dispatchers、CoroutineDispatcher、ContinuationInterceptor、CoroutineContext之间的关系
public actual object Dispatchers {
@JvmStatic
public actual val Default: CoroutineDispatcher = DefaultScheduler
@JvmStatic
public actual val Main: MainCoroutineDispatcher get() = MainDispatcherLoader.dispatcher
@JvmStatic
public actual val Unconfined: CoroutineDispatcher = kotlinx.coroutines.Unconfined
@JvmStatic
public val IO: CoroutineDispatcher = DefaultIoScheduler
}
public abstract class CoroutineDispatcher :
AbstractCoroutineContextElement(ContinuationInterceptor), ContinuationInterceptor {
}
public interface ContinuationInterceptor : CoroutineContext.Element {}
public interface Element : CoroutineContext {}
Dispatchers中存放的是协程调度器（它本身是一个单例），有我们平时常用的IO、Default、Main等。这些协程调度器都是CoroutineDispatcher的子类，这些协程调度器其实都是CoroutineContext。
demo
我们先来看一个关于launch的demo：
fun main() {
val coroutineScope = CoroutineScope(Job())
coroutineScope.launch {
println("Thread : ${Thread.currentThread().name}")
}
Thread.sleep(5000L)
}
在生成CoroutineScope时，demo中没有传入相关的协程调度器，也就是Dispatchers。那这个launch会运行到哪个线程之上？
运行试一下：
Thread : DefaultDispatcher-worker-1
居然运行到了DefaultDispatcher-worker-1线程上，这看起来明显是Dispatchers.Default协程调度器里面的线程。我明明没传Dispatchers相关的context，居然会运行到子线程上。说明运行到default线程是launch默认的。
它是怎么与default线程产生关联的？打开源码一探究竟：
public fun CoroutineScope.launch(
context: CoroutineContext = EmptyCoroutineContext,
start: CoroutineStart = CoroutineStart.DEFAULT,
block: suspend CoroutineScope.() -> Unit
): Job {
//代码1
val newContext = newCoroutineContext(context)
//代码2
val coroutine = if (start.isLazy)
LazyStandaloneCoroutine(newContext, block) else
StandaloneCoroutine(newContext, active = true)
//代码3
coroutine.start(start, coroutine, block)
return coroutine
}
- 将传入的CoroutineContext构造出新的context
- 启动模式，判断是否为懒加载，如果是懒加载则构建懒加载协程对象，否则就是标准的
- 启动协程
我们重点关注代码1，这是与CoroutineContext相关的。
public actual fun CoroutineScope.newCoroutineContext(context: CoroutineContext): CoroutineContext {
//从父协程那里继承过来的context+这次的context
val combined = coroutineContext.foldCopiesForChildCoroutine() + context
val debug = if (DEBUG) combined + CoroutineId(COROUTINE_ID.incrementAndGet()) else combined
//combined可以简单的把它看成是一个map，它是CoroutineContext类型的
//如果当前context不等于Dispatchers.Default，而且从map里面取ContinuationInterceptor（用于拦截之后分发线程的）值为空，说明没有传入协程应该在哪个线程上运行的相关参数
return if (combined !== Dispatchers.Default && combined[ContinuationInterceptor] == null)
debug + Dispatchers.Default else debug
}
调用launch的时候，我们没有传入context，默认参数是EmptyCoroutineContext。这里的combined，它其实是CoroutineContext类型的，可以简单的看成是map（其实不是，只是类似）。通过combined[ContinuationInterceptor]可以将传入的线程调度相关的参数给取出来，这里如果取出来为空，是给该context添加了一个Dispatchers.Default，然后把新的context返回出去了。所以launch默认情况下，会走到default线程去执行。
补充一点：CoroutineContext能够通过+连接是因为它内部有个public operator fun plus函数。能够通过combined[ContinuationInterceptor]这种方式访问元素是因为有个public operator fun get函数。
public interface CoroutineContext {
/**
* Returns the element with the given [key] from this context or `null`.
*/
public operator fun <E : Element> get(key: Key<E>): E?
/**
* Returns a context containing elements from this context and elements from other [context].
* The elements from this context with the same key as in the other one are dropped.
*/
public operator fun plus(context: CoroutineContext): CoroutineContext {
......
}
}
startCoroutineCancellable
上面我们分析了launch默认情况下，context中会增加Dispatchers.Default的这个协程调度器，到时launch的Lambda会在default线程上执行，其中具体流程是怎么样的，我们分析一下。
在之前的文章 Kotlin协程之launch原理 中我们分析过，launch默认情况下会最终执行到startCoroutineCancellable函数。
public fun <T> (suspend () -> T).startCoroutineCancellable(completion: Continuation<T>): Unit = runSafely(completion) {
//构建ContinuationImpl
createCoroutineUnintercepted(completion).intercepted().resumeCancellableWith(Result.success(Unit))
}
public actual fun <T> (suspend () -> T).createCoroutineUnintercepted(
completion: Continuation<T>
): Continuation<Unit> {
val probeCompletion = probeCoroutineCreated(completion)
return if (this is BaseContinuationImpl)
//走这里
create(probeCompletion)
else
createCoroutineFromSuspendFunction(probeCompletion) {
(this as Function1<Continuation<T>, Any?>).invoke(it)
}
}
在Kotlin协程之launch原理 文章中，咱们分析过create(probeCompletion)这里创建出来的是launch的那个Lambda，编译器会产生一个匿名内部类，它继承自SuspendLambda，而SuspendLambda是继承自ContinuationImpl。所以 createCoroutineUnintercepted(completion)一开始构建出来的是一个ContinuationImpl，接下来需要去看它的intercepted()函数。
internal abstract class ContinuationImpl(
completion: Continuation<Any?>?,
private val _context: CoroutineContext?
) : BaseContinuationImpl(completion) {
constructor(completion: Continuation<Any?>?) : this(completion, completion?.context)
public override val context: CoroutineContext
get() = _context!!
@Transient
private var intercepted: Continuation<Any?>? = null
public fun intercepted(): Continuation<Any?> =
intercepted
?: (context[ContinuationInterceptor]?.interceptContinuation(this) ?: this)
.also { intercepted = it }
}
第一次走到intercepted()函数时，intercepted肯定是为null的，还没初始化。此时会通过context[ContinuationInterceptor]取出Dispatcher对象，然后调用该Dispatcher对象的interceptContinuation()函数。这个Dispatcher对象在demo这里其实就是Dispatchers.Default。
public actual object Dispatchers {
@JvmStatic
public actual val Default: CoroutineDispatcher = DefaultScheduler
}
可以看到，Dispatchers.Default是一个CoroutineDispatcher对象，interceptContinuation()函数就在CoroutineDispatcher中。
public abstract class CoroutineDispatcher :
AbstractCoroutineContextElement(ContinuationInterceptor), ContinuationInterceptor {
public final override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> =
DispatchedContinuation(this, continuation)
}
public fun <T> (suspend () -> T).startCoroutineCancellable(completion: Continuation<T>): Unit = runSafely(completion) {
createCoroutineUnintercepted(completion).intercepted().resumeCancellableWith(Result.success(Unit))
}
这个方法非常简单，就是新建并且返回了一个DispatchedContinuation对象，将this和continuation给传入进去。这里的this是Dispatchers.Default。
所以，最终我们发现走完startCoroutineCancellable的前2步之后，也就是走完intercepted()之后，创建的是DispatchedContinuation对象，最后是调用的DispatchedContinuation的resumeCancellableWith函数。最后这步比较关键，这是真正将协程的具体执行逻辑放到线程上执行的部分。
internal class DispatchedContinuation<in T>(
//这里传入的dispatcher在demo中是Dispatchers.Default
@JvmField val dispatcher: CoroutineDispatcher,
@JvmField val continuation: Continuation<T>
) : DispatchedTask<T>(MODE_UNINITIALIZED), CoroutineStackFrame, Continuation<T> by continuation {
inline fun resumeCancellableWith(
result: Result<T>,
noinline onCancellation: ((cause: Throwable) -> Unit)?
) {
val state = result.toState(onCancellation)
//代码1
if (dispatcher.isDispatchNeeded(context)) {
_state = state
resumeMode = MODE_CANCELLABLE
//代码2
dispatcher.dispatch(context, this)
} else {
//代码3
executeUnconfined(state, MODE_CANCELLABLE) {
if (!resumeCancelled(state)) {
resumeUndispatchedWith(result)
}
}
}
}
}
internal abstract class DispatchedTask<in T>(
@JvmField public var resumeMode: Int
) : SchedulerTask() {
......
}
internal actual typealias SchedulerTask = Task
internal abstract class Task(
@JvmField var submissionTime: Long,
@JvmField var taskContext: TaskContext
) : Runnable {
......
}
public abstract class CoroutineDispatcher :
AbstractCoroutineContextElement(ContinuationInterceptor), ContinuationInterceptor {
public abstract fun dispatch(context: CoroutineContext, block: Runnable)
public open fun isDispatchNeeded(context: CoroutineContext): Boolean = true
}
从DispatchedContinuation的继承结构来看，它既是一个Continuation（通过委托给传入的continuation参数），也是一个Runnable。
- 首先看代码1：这个dispatcher在demo中其实是Dispatchers.Default ，然后调用它的isDispatchNeeded(),这个函数定义在CoroutineDispatcher中，默认就是返回true，只有Dispatchers.Unconfined返回false
- 代码2：调用Dispatchers.Default的dispatch函数，将context和自己（DispatchedContinuation,也就是Runnable）传过去了
- 代码3：对应Dispatchers.Unconfined的情况，它的isDispatchNeeded()返回false
现在我们要分析代码2之后的执行逻辑，也就是将context和Runnable传入到dispatch函数之后是怎么执行的。按道理，看到Runnable，那可能这个与线程执行相关，应该离我们想要的答案不远了。回到Dispatchers，我们发现Dispatchers.Default是DefaultScheduler类型的，那我们就去DefaultScheduler中或者其父类中去找dispatch函数。
public actual object Dispatchers {
@JvmStatic
public actual val Default: CoroutineDispatcher = DefaultScheduler
}
internal object DefaultScheduler : SchedulerCoroutineDispatcher(
CORE_POOL_SIZE, MAX_POOL_SIZE,
IDLE_WORKER_KEEP_ALIVE_NS, DEFAULT_SCHEDULER_NAME
) {
......
}
internal open class SchedulerCoroutineDispatcher(
private val corePoolSize: Int = CORE_POOL_SIZE,
private val maxPoolSize: Int = MAX_POOL_SIZE,
private val idleWorkerKeepAliveNs: Long = IDLE_WORKER_KEEP_ALIVE_NS,
private val schedulerName: String = "CoroutineScheduler",
) : ExecutorCoroutineDispatcher() {
private var coroutineScheduler = createScheduler()
private fun createScheduler() =
CoroutineScheduler(corePoolSize, maxPoolSize, idleWorkerKeepAliveNs, schedulerName)
override fun dispatch(context: CoroutineContext, block: Runnable): Unit = coroutineScheduler.dispatch(block)
}
最后发现dispatch函数在其父类SchedulerCoroutineDispatcher中，在这里构建了一个CoroutineScheduler，直接调用了CoroutineScheduler对象的dispatch，然后将Runnable（也就是上面的DispatchedContinuation对象）传入。
internal class CoroutineScheduler(
@JvmField val corePoolSize: Int,
@JvmField val maxPoolSize: Int,
@JvmField val idleWorkerKeepAliveNs: Long = IDLE_WORKER_KEEP_ALIVE_NS,
@JvmField val schedulerName: String = DEFAULT_SCHEDULER_NAME
) : Executor, Closeable {
override fun execute(command: Runnable) = dispatch(command)
fun dispatch(block: Runnable, taskContext: TaskContext = NonBlockingContext, tailDispatch: Boolean = false) {
trackTask() // this is needed for virtual time support
//代码1：构建Task，Task实现了Runnable接口
val task = createTask(block, taskContext)
//代码2：取当前线程转为Worker对象,Worker是一个继承自Thread的类
val currentWorker = currentWorker()
//代码3：尝试将Task提交到本地队列并根据结果执行相应的操作
val notAdded = currentWorker.submitToLocalQueue(task, tailDispatch)
if (notAdded != null) {
//代码4：notAdded不为null，则再将notAdded（Task）添加到全局队列中
if (!addToGlobalQueue(notAdded)) {
throw RejectedExecutionException("$schedulerName was terminated")
}
}
val skipUnpark = tailDispatch && currentWorker != null
// Checking 'task' instead of 'notAdded' is completely okay
if (task.mode == TASK_NON_BLOCKING) {
if (skipUnpark) return
//代码5: 创建Worker并开始执行该线程
signalCpuWork()
} else {
// Increment blocking tasks anyway
signalBlockingWork(skipUnpark = skipUnpark)
}
}
private fun currentWorker(): Worker? = (Thread.currentThread() as? Worker)?.takeIf { it.scheduler == this }
internal inner class Worker private constructor() : Thread() {
.....
}
}
观察发现，原来CoroutineScheduler类实现了java.util.concurrent.Executor接口，同时实现了它的execute方法，这个方法也会调用dispatch()。
- 代码1：首先是通过Runnable构建了一个Task，这个Task其实也是实现了Runnable接口，只是把传入的Runnable包装了一下
- 代码2：将当前线程取出来转换成Worker，当然第一次时，这个转换不会成功，这个Worker是继承自Thread的一个类
- 代码3：将task提交到本地队列中，这个本地队列待会儿会在Worker这个线程执行时取出Task，并执行Task
- 代码4：如果task提交到本地队列的过程中没有成功，那么会添加到全局队列中，待会儿也会被Worker取出来Task并执行
- 代码5：创建Worker线程，并开始执行
开始执行Worker线程之后，我们需要看一下这个线程的run方法执行的是啥，也就是它的具体执行逻辑。
internal inner class Worker private constructor() : Thread() {
override fun run() = runWorker()
private fun runWorker() {
var rescanned = false
while (!isTerminated && state != WorkerState.TERMINATED) {
//代码1
val task = findTask(mayHaveLocalTasks)
if (task != null) {
rescanned = false
minDelayUntilStealableTaskNs = 0L
//代码2
executeTask(task)
continue
} else {
mayHaveLocalTasks = false
}
if (minDelayUntilStealableTaskNs != 0L) {
if (!rescanned) {
rescanned = true
} else {
rescanned = false
tryReleaseCpu(WorkerState.PARKING)
interrupted()
LockSupport.parkNanos(minDelayUntilStealableTaskNs)
minDelayUntilStealableTaskNs = 0L
}
continue
}
tryPark()
}
tryReleaseCpu(WorkerState.TERMINATED)
}
fun findTask(scanLocalQueue: Boolean): Task? {
if (tryAcquireCpuPermit()) return findAnyTask(scanLocalQueue)
// If we can't acquire a CPU permit -- attempt to find blocking task
val task = if (scanLocalQueue) {
localQueue.poll() ?: globalBlockingQueue.removeFirstOrNull()
} else {
globalBlockingQueue.removeFirstOrNull()
}
return task ?: trySteal(blockingOnly = true)
}
private fun executeTask(task: Task) {
val taskMode = task.mode
idleReset(taskMode)
beforeTask(taskMode)
runSafely(task)
afterTask(taskMode)
}
fun runSafely(task: Task) {
try {
task.run()
} catch (e: Throwable) {
val thread = Thread.currentThread()
thread.uncaughtExceptionHandler.uncaughtException(thread, e)
} finally {
unTrackTask()
}
}
}
run方法直接调用的runWorker()，在里面是一个while循环，不断从队列中取Task来执行。
- 代码1：从本地队列或者全局队列中取出Task
- 代码2：执行这个task，最终其实就是调用这个Runnable的run方法。
也就是说，在Worker这个线程中，执行了这个Runnable的run方法。还记得这个Runnable是谁么？它就是上面我们看过的DispatchedContinuation，这里的run方法执行的就是协程任务，那这块具体的run方法的实现逻辑，我们应该到DispatchedContinuation中去找。
internal class DispatchedContinuation<in T>(
@JvmField val dispatcher: CoroutineDispatcher,
@JvmField val continuation: Continuation<T>
) : DispatchedTask<T>(MODE_UNINITIALIZED), CoroutineStackFrame, Continuation<T> by continuation {
......
}
internal abstract class DispatchedTask<in T>(
@JvmField public var resumeMode: Int
) : SchedulerTask() {
public final override fun run() {
assert { resumeMode != MODE_UNINITIALIZED } // should have been set before dispatching
val taskContext = this.taskContext
var fatalException: Throwable? = null
try {
val delegate = delegate as DispatchedContinuation<T>
val continuation = delegate.continuation
withContinuationContext(continuation, delegate.countOrElement) {
val context = continuation.context
val state = takeState() // NOTE: Must take state in any case, even if cancelled
val exception = getExceptionalResult(state)
/*
* Check whether continuation was originally resumed with an exception.
* If so, it dominates cancellation, otherwise the original exception
* will be silently lost.
*/
val job = if (exception == null && resumeMode.isCancellableMode) context[Job] else null
//非空，且未处于active状态
if (job != null && !job.isActive) {
//开始之前，协程已经被取消，将具体的Exception传出去
val cause = job.getCancellationException()
cancelCompletedResult(state, cause)
continuation.resumeWithStackTrace(cause)
} else {
//有异常，传递异常
if (exception != null) {
continuation.resumeWithException(exception)
} else {
//代码1
continuation.resume(getSuccessfulResult(state))
}
}
}
} catch (e: Throwable) {
// This instead of runCatching to have nicer stacktrace and debug experience
fatalException = e
} finally {
val result = runCatching { taskContext.afterTask() }
handleFatalException(fatalException, result.exceptionOrNull())
}
}
}
我们主要看一下代码1处，调用了resume开启协程。前面没有异常，才开始启动协程，这里才是真正的开始启动协程，开始执行launch传入的Lambda表达式。这个时候，协程的逻辑是在Worker这个线程上执行的了，切到某个线程上执行的逻辑已经完成了。
ps: rusume会走到BaseContinuationImpl的rusumeWith，然后走到launch传入的Lambda匿名内部类的invokeSuspend方法，开始执行状态机逻辑。前面的文章 Kotlin协程createCoroutine和startCoroutine原理 我们分析过这里，这里就只是简单提一下。
到这里，Dispatchers的执行流程就算完了，前后都串起来了。
小结
Dispatchers是协程框架中与线程交互的关键。底层会有不同的线程池，Dispatchers.Default、IO，协程任务来了的时候会封装成一个个的Runnable，丢到线程中执行，这些Runnable的run方法中执行的其实就是continuation.resume，也就是launch的Lambda生成的SuspendLambda匿名内部类，也就是开启协程状态机，开始协程的真正执行。
作者：潇风寒月
链接：https://juejin.cn/post/7127492385923137549
来源：稀土掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
