Kotlin協(xié)程不是什么空中閣樓舰罚，Kotlin源代碼會被編譯成class字節(jié)碼文件方援，最終會運行到虛擬機中黎休。所以從本質(zhì)上講，Kotlin和Java是類似的陶耍，都是可以編譯產(chǎn)生class的語言苦丁，但最終還是會受到虛擬機的限制，它們的代碼最終會在虛擬機上的某個線程上被執(zhí)行物臂。
之前我們分析了launch的原理，但當(dāng)時我們沒有去分析協(xié)程創(chuàng)建出來后是如何與線程產(chǎn)生關(guān)聯(lián)的产上，怎么被分發(fā)到具體的線程上執(zhí)行的棵磷，本篇文章就帶大家分析一下。
前置知識
要想搞懂Dispatchers晋涣，我們先來看一下Dispatchers仪媒、CoroutineDispatcher、ContinuationInterceptor谢鹊、CoroutineContext之間的關(guān)系
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中存放的是協(xié)程調(diào)度器（它本身是一個單例）算吩，有我們平時常用的IO、Default佃扼、Main等偎巢。這些協(xié)程調(diào)度器都是CoroutineDispatcher的子類，這些協(xié)程調(diào)度器其實都是CoroutineContext兼耀。
demo
我們先來看一個關(guān)于launch的demo：
fun main() {
val coroutineScope = CoroutineScope(Job())
coroutineScope.launch {
println("Thread : ${Thread.currentThread().name}")
}
Thread.sleep(5000L)
}
在生成CoroutineScope時压昼，demo中沒有傳入相關(guān)的協(xié)程調(diào)度器求冷，也就是Dispatchers。那這個launch會運行到哪個線程之上窍霞？
運行試一下：
Thread : DefaultDispatcher-worker-1
居然運行到了DefaultDispatcher-worker-1線程上匠题，這看起來明顯是Dispatchers.Default協(xié)程調(diào)度器里面的線程。我明明沒傳Dispatchers相關(guān)的context但金，居然會運行到子線程上韭山。說明運行到default線程是launch默認(rèn)的。
它是怎么與default線程產(chǎn)生關(guān)聯(lián)的冷溃？打開源碼一探究竟：
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構(gòu)造出新的context
- 啟動模式钱磅，判斷是否為懶加載，如果是懶加載則構(gòu)建懶加載協(xié)程對象秃诵，否則就是標(biāo)準(zhǔn)的
- 啟動協(xié)程
我們重點關(guān)注代碼1续搀，這是與CoroutineContext相關(guān)的。
public actual fun CoroutineScope.newCoroutineContext(context: CoroutineContext): CoroutineContext {
//從父協(xié)程那里繼承過來的context+這次的context
val combined = coroutineContext.foldCopiesForChildCoroutine() + context
val debug = if (DEBUG) combined + CoroutineId(COROUTINE_ID.incrementAndGet()) else combined
//combined可以簡單的把它看成是一個map菠净，它是CoroutineContext類型的
//如果當(dāng)前context不等于Dispatchers.Default禁舷，而且從map里面取ContinuationInterceptor（用于攔截之后分發(fā)線程的）值為空，說明沒有傳入?yún)f(xié)程應(yīng)該在哪個線程上運行的相關(guān)參數(shù)
return if (combined !== Dispatchers.Default && combined[ContinuationInterceptor] == null)
debug + Dispatchers.Default else debug
}
調(diào)用launch的時候毅往，我們沒有傳入context牵咙，默認(rèn)參數(shù)是EmptyCoroutineContext。這里的combined攀唯，它其實是CoroutineContext類型的洁桌，可以簡單的看成是map（其實不是，只是類似）侯嘀。通過combined[ContinuationInterceptor]可以將傳入的線程調(diào)度相關(guān)的參數(shù)給取出來另凌，這里如果取出來為空，是給該context添加了一個Dispatchers.Default戒幔，然后把新的context返回出去了吠谢。所以launch默認(rèn)情況下，會走到default線程去執(zhí)行诗茎。
補充一點：CoroutineContext能夠通過+連接是因為它內(nèi)部有個public operator fun plus函數(shù)工坊。能夠通過combined[ContinuationInterceptor]這種方式訪問元素是因為有個public operator fun get函數(shù)。
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默認(rèn)情況下敢订，context中會增加Dispatchers.Default的這個協(xié)程調(diào)度器王污，到時launch的Lambda會在default線程上執(zhí)行，其中具體流程是怎么樣的楚午，我們分析一下昭齐。
在之前的文章 Kotlin協(xié)程之launch原理 中我們分析過，launch默認(rèn)情況下會最終執(zhí)行到startCoroutineCancellable函數(shù)矾柜。
public fun <T> (suspend () -> T).startCoroutineCancellable(completion: Continuation<T>): Unit = runSafely(completion) {
//構(gòu)建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協(xié)程之launch原理 文章中司浪，咱們分析過create(probeCompletion)這里創(chuàng)建出來的是launch的那個Lambda泊业，編譯器會產(chǎn)生一個匿名內(nèi)部類，它繼承自SuspendLambda啊易，而SuspendLambda是繼承自ContinuationImpl吁伺。所以 createCoroutineUnintercepted(completion)一開始構(gòu)建出來的是一個ContinuationImpl，接下來需要去看它的intercepted()函數(shù)租谈。
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()函數(shù)時篮奄，intercepted肯定是為null的，還沒初始化割去。此時會通過context[ContinuationInterceptor]取出Dispatcher對象窟却，然后調(diào)用該Dispatcher對象的interceptContinuation()函數(shù)。這個Dispatcher對象在demo這里其實就是Dispatchers.Default呻逆。
public actual object Dispatchers {
@JvmStatic
public actual val Default: CoroutineDispatcher = DefaultScheduler
}
可以看到夸赫，Dispatchers.Default是一個CoroutineDispatcher對象，interceptContinuation()函數(shù)就在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給傳入進(jìn)去宜雀。這里的this是Dispatchers.Default切平。
所以，最終我們發(fā)現(xiàn)走完startCoroutineCancellable的前2步之后辐董，也就是走完intercepted()之后悴品，創(chuàng)建的是DispatchedContinuation對象，最后是調(diào)用的DispatchedContinuation的resumeCancellableWith函數(shù)简烘。最后這步比較關(guān)鍵苔严，這是真正將協(xié)程的具體執(zhí)行邏輯放到線程上執(zhí)行的部分。
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的繼承結(jié)構(gòu)來看孤澎，它既是一個Continuation（通過委托給傳入的continuation參數(shù)）邦蜜，也是一個Runnable。
- 首先看代碼1：這個dispatcher在demo中其實是Dispatchers.Default 亥至，然后調(diào)用它的isDispatchNeeded(),這個函數(shù)定義在CoroutineDispatcher中，默認(rèn)就是返回true贱迟，只有Dispatchers.Unconfined返回false
- 代碼2：調(diào)用Dispatchers.Default的dispatch函數(shù)姐扮，將context和自己（DispatchedContinuation,也就是Runnable）傳過去了
- 代碼3：對應(yīng)Dispatchers.Unconfined的情況，它的isDispatchNeeded()返回false
現(xiàn)在我們要分析代碼2之后的執(zhí)行邏輯衣吠，也就是將context和Runnable傳入到dispatch函數(shù)之后是怎么執(zhí)行的茶敏。按道理，看到Runnable缚俏，那可能這個與線程執(zhí)行相關(guān)惊搏，應(yīng)該離我們想要的答案不遠(yuǎn)了贮乳。回到Dispatchers恬惯，我們發(fā)現(xiàn)Dispatchers.Default是DefaultScheduler類型的向拆，那我們就去DefaultScheduler中或者其父類中去找dispatch函數(shù)。
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)
}
最后發(fā)現(xiàn)dispatch函數(shù)在其父類SchedulerCoroutineDispatcher中酪耳，在這里構(gòu)建了一個CoroutineScheduler浓恳，直接調(diào)用了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：構(gòu)建Task颈将，Task實現(xiàn)了Runnable接口
val task = createTask(block, taskContext)
//代碼2：取當(dāng)前線程轉(zhuǎn)為Worker對象,Worker是一個繼承自Thread的類
val currentWorker = currentWorker()
//代碼3：嘗試將Task提交到本地隊列并根據(jù)結(jié)果執(zhí)行相應(yīng)的操作
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: 創(chuàng)建Worker并開始執(zhí)行該線程
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() {
.....
}
}
觀察發(fā)現(xiàn)言疗，原來CoroutineScheduler類實現(xiàn)了java.util.concurrent.Executor接口晴圾，同時實現(xiàn)了它的execute方法，這個方法也會調(diào)用dispatch()噪奄。
- 代碼1：首先是通過Runnable構(gòu)建了一個Task死姚，這個Task其實也是實現(xiàn)了Runnable接口，只是把傳入的Runnable包裝了一下
- 代碼2：將當(dāng)前線程取出來轉(zhuǎn)換成Worker梗醇，當(dāng)然第一次時知允，這個轉(zhuǎn)換不會成功，這個Worker是繼承自Thread的一個類
- 代碼3：將task提交到本地隊列中叙谨，這個本地隊列待會兒會在Worker這個線程執(zhí)行時取出Task温鸽，并執(zhí)行Task
- 代碼4：如果task提交到本地隊列的過程中沒有成功，那么會添加到全局隊列中手负，待會兒也會被Worker取出來Task并執(zhí)行
- 代碼5：創(chuàng)建Worker線程涤垫，并開始執(zhí)行
開始執(zhí)行Worker線程之后，我們需要看一下這個線程的run方法執(zhí)行的是啥竟终，也就是它的具體執(zhí)行邏輯蝠猬。
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方法直接調(diào)用的runWorker()，在里面是一個while循環(huán)统捶，不斷從隊列中取Task來執(zhí)行榆芦。
- 代碼1：從本地隊列或者全局隊列中取出Task
- 代碼2：執(zhí)行這個task，最終其實就是調(diào)用這個Runnable的run方法喘鸟。
也就是說匆绣，在Worker這個線程中，執(zhí)行了這個Runnable的run方法什黑。還記得這個Runnable是誰么崎淳？它就是上面我們看過的DispatchedContinuation，這里的run方法執(zhí)行的就是協(xié)程任務(wù)愕把，那這塊具體的run方法的實現(xiàn)邏輯拣凹，我們應(yīng)該到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狀態(tài)
if (job != null && !job.isActive) {
//開始之前嚣镜，協(xié)程已經(jīng)被取消爬迟，將具體的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處祈惶，調(diào)用了resume開啟協(xié)程雕旨。前面沒有異常，才開始啟動協(xié)程捧请，這里才是真正的開始啟動協(xié)程凡涩，開始執(zhí)行l(wèi)aunch傳入的Lambda表達(dá)式。這個時候疹蛉，協(xié)程的邏輯是在Worker這個線程上執(zhí)行的了活箕，切到某個線程上執(zhí)行的邏輯已經(jīng)完成了。
ps: rusume會走到BaseContinuationImpl的rusumeWith可款，然后走到launch傳入的Lambda匿名內(nèi)部類的invokeSuspend方法育韩，開始執(zhí)行狀態(tài)機邏輯。前面的文章 Kotlin協(xié)程createCoroutine和startCoroutine原理 我們分析過這里闺鲸，這里就只是簡單提一下筋讨。
到這里，Dispatchers的執(zhí)行流程就算完了摸恍，前后都串起來了悉罕。
小結(jié)
Dispatchers是協(xié)程框架中與線程交互的關(guān)鍵。底層會有不同的線程池立镶，Dispatchers.Default壁袄、IO，協(xié)程任務(wù)來了的時候會封裝成一個個的Runnable媚媒，丟到線程中執(zhí)行嗜逻，這些Runnable的run方法中執(zhí)行的其實就是continuation.resume，也就是launch的Lambda生成的SuspendLambda匿名內(nèi)部類缭召，也就是開啟協(xié)程狀態(tài)機栈顷，開始協(xié)程的真正執(zhí)行。
