網站首頁編程語言正文

Kotlin協程的線程調度示例詳解_Android

作者：rencai ? 更新時間： 2023-01-09 編程語言

引言

在第一篇文章中我們分析了協程啟動創建過程啟動過程，在本文中，我們將著重剖析協程中協程調度的邏輯流程。主要是分析解答如下2個問題：

涉及到協程方法器是如何將協程代碼調度到特定的線程執行？
子協程執行完又是如何切換0回父協程的線程環境？

一、協程的分發器作用

1.1 測試代碼

GlobalScope.launch {
    //協程體1
    Log.d(TAG, "before suspend job.")
    withContext(Dispatchers.Main) {
        //協程體2
        Log.d(TAG, "print in Main thread.")
    }
    Log.d(TAG, "after suspend job.")
}

此次的協程測試用例中，我們默認的launch一個協程，我們簡單的將launch需要執行的這外層邏輯為協程體1。
在協程體1中，我們使用withContext將協程切換到主線程執行，打印日志。我們將這里面執行的協程邏輯為協程體2。
協程體2執行完成后，切回協程體1中執行并打印Log。
注意，根據我們之前《協程的創建與啟動》文章中分析的，Kotlin編譯器針對協程體1和協程體2分別生成一個繼承與SuspenLamabda的類型，比如：class MainActivity#onCreate$1 : SuspenLambda{...}。我們在講協程體時，也同時代指這個類實例。

繼續跟蹤launch()函數執行邏輯，這次跟蹤過程不同與《協程的創建與啟動》篇章，我們會將側重點放在啟動過程中協程調度器是如何起作用的？接下來見1.2

1.2 CoroutineScope.launch

public fun CoroutineScope.launch(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> Unit
): Job {
    //1. 見1.2.1
    val newContext = newCoroutineContext(context)
    val coroutine = if (start.isLazy)
        LazyStandaloneCoroutine(newContext, block) else
        StandaloneCoroutine(newContext, active = true)
    //2. 詳見1.3
    coroutine.start(start, coroutine, block)
    return coroutine
}

這里會新建一個CoroutineContext，詳見1.2.1
根據之前的分析，這個里最終會調用到startCoroutineCancellable()方法，詳見1.3流程。

1.2.1 newCoroutineContext

public actual fun CoroutineScope.newCoroutineContext(context: CoroutineContext): CoroutineContext {
    val combined = foldCopies(coroutineContext, context, true)
    val debug = if (DEBUG) combined + CoroutineId(COROUTINE_ID.incrementAndGet()) else combined
    return 
    if (combined !== Dispatchers.Default && combined[ContinuationInterceptor] == null)
        debug + Dispatchers.Default
    else 
    	debug
}

coroutineContext：coroutineContext是CoroutineScope的成員變量，當此時為GlobalScope.coroutineContext==EmptyCoroutineContext

context：由于調用launch時沒有指定Context，所以傳到此處也是EmptyCoroutineContext。foldCopies()函數將2個context相加并拷貝，最終combied==EmptyCoroutineContext。

而在return這最后判斷返回的是debug+Dispatchers.Defatult，所以此時默認的分發器為Dispatchers.Defatult。

這里涉及到的協程Context運算不做深入剖析，簡單可以認為協程重寫了“+”運算，使得Context之間可以使用“+”來疊加，沒有的Element類型會被添加到Element集合，集合中已有的Element類型會被覆蓋。

1.3 startCoroutineCancellable

internal fun <R, T> (suspend (R) -> T).startCoroutineCancellable(
    receiver: R, completion: Continuation<T>,
    onCancellation: ((cause: Throwable) -> Unit)? = null
) =
    runSafely(completion) {
    	//1. 創建SuspendLambda協程體
        createCoroutineUnintercepted(receiver, completion)
            //2. 攔截：取出分發器，并構建方法器Continuation。詳見1.3.1
            .intercepted()
            //3. 調用方法器Continuation的resume方法，詳見1.4
            .resumeCancellableWith(Result.success(Unit), onCancellation)
    }

這里的構建協程體在《協程的創建與啟動》一節中已經剖析，不再贅述。
進行攔截，注意：這里其實會根據方法器再構建出一個DispatchedContinuation對象，它也是一個續體類型，這是對協程體的一次包裝。詳見1.3.1小節。
調用攔截器續體的resumeCancellableWith()開始狀態機流轉，執行分發流程詳見1.4小節。

1.3.1 intercepted()

 public fun intercepted(): Continuation<Any?> =
        intercepted?: (
                //1. 取出攔截器
                context[ContinuationInterceptor]?
                    //2.構建攔截器續體
                    .interceptContinuation(this)?: this)
                .also { intercepted = it }

取出當前上下文中的攔截器類型，根據之前1.2.1小節的分析，這里取出來的是Dispatchers.Defatult。
interceptContinuation(this)為構建攔截器續體，注意這里傳入的this是協程體1。 詳見1.3.2。

1.3.2 CoroutineDispatcher

//Base class to be extended by all coroutine dispatcher implementations.
public abstract class CoroutineDispatcher :
    AbstractCoroutineContextElement(ContinuationInterceptor), ContinuationInterceptor {
public final override fun <T> interceptContinuation(continuation: Continuation<T>):
        //詳見1.4
        Continuation<T> = DispatchedContinuation(this, continuation)
}

直接新建了一個DispatchedContinuation對象實例這里需要注意傳入的構建參數：

this：當前Dispatcher，也就是Dispatchers.Defatult。
continuation：協程體1。

1.3.3 小結

自此Continuation.intercepted()方法就分析結束，最終的結果是：用上下文中的Dispatcher和當前Contination對象也就是協程體1，共同作為構建參數，新建了一個DispatchedContinuation對象。

接下來接著1.3中的第三點，調用DispatchedContinuation.resumeCancellableWith()方法開始分析。

1.4 DispatchedContinuation

internal class DispatchedContinuation<in T>(
    //1. 分發器
    @JvmField val dispatcher: CoroutineDispatcher,
	//2. 注意這里將Continuation的實現委托給了continuation成員變量。
    @JvmField val continuation: Continuation<T>
) : DispatchedTask<T>(MODE_UNINITIALIZED)
, CoroutineStackFrame,
Continuation<T> by continuation {
    	//3. 復寫屬性delegate為自己
	    override val delegate: Continuation<T>
        get() = this
    ...
    // We inline it to save an entry on the stack in cases where it shows (unconfined dispatcher)
    // It is used only in Continuation<T>.resumeCancellableWith
    @Suppress("NOTHING_TO_INLINE")
    inline fun resumeCancellableWith(
        result: Result<T>,
        noinline onCancellation: ((cause: Throwable) -> Unit)?
    ) {
        val state = result.toState(onCancellation)
        //默認為true
        if (dispatcher.isDispatchNeeded(context)) {
            _state = state
            resumeMode = MODE_CANCELLABLE
            //4. 詳細見
            dispatcher.dispatch(context, this)
        } else {
            executeUnconfined(state, MODE_CANCELLABLE) {
                if (!resumeCancelled(state)) {
                    resumeUndispatchedWith(result)
                }
            }
        }
    }
}

這里的dispatcher==Dispatchers.Defatult，所以接下來需要解析Dispatchers.Defatult到底是什么東西。詳見1.5

成員變量dispatcher==Dispatchers.Default。
成員變量continucation==協程體1（SuspenLambda類型實例）。同時DispatchedContinuation繼承于Continuation接口，它將Continuation接口的實現委托給了成員變量continuation。
deleagte為復寫了DispatchedTask.delegate屬性，將其返回自己。
調用分發器也就是Dispatchers.Defatult的dispatch()方法，注意這里傳入的參數：

context：來自Continuation接口的屬性，由于委托給了成員變量continuation，所以此context==continuation.context。

this：分發器本身Dispatchers.Defatult

自此這個方法的分析結束：調用分發器的進行分發，接下來分析就開始分析協程方法器CoroutineDispatcher

1.5 DefaultScheduler

//Dispathcer.kt
@JvmStatic
public actual val Default: CoroutineDispatcher = DefaultScheduler
//Dispathcer.kt
// Instance of Dispatchers.Default
internal object DefaultScheduler : SchedulerCoroutineDispatcher(
    CORE_POOL_SIZE, MAX_POOL_SIZE,
    IDLE_WORKER_KEEP_ALIVE_NS, DEFAULT_SCHEDULER_NAME
) {
    ...
}

實際上是繼承 SchedulerCoroutineDispatcher類型。詳見1.5.1

1.5.1 SchedulerCoroutineDispatcher

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() {
    override val executor: Executor
        get() = coroutineScheduler
    // This is variable for test purposes, so that we can reinitialize from clean state
    private var coroutineScheduler = createScheduler()
    private fun createScheduler() =
        //1. 詳見1.5.2
        CoroutineScheduler(corePoolSize, maxPoolSize, idleWorkerKeepAliveNs, schedulerName)
    //2. 詳見1.5.2
    override fun dispatch(context: CoroutineContext, block: Runnable): Unit 
    = coroutineScheduler.dispatch(block)
    ...
}
//Executors.kt
//2. 實際上是繼承ExecutorCoroutineDispatcher
public abstract class ExecutorCoroutineDispatcher: CoroutineDispatcher(), Closeable {
    ...
}

可以看到實際上調用了CoroutineScheduler.dispatch方法。此時發現，第二個參數是Runnable類型的，而在1.4小節中，我們知道傳入的是this也就是DispatchedContinuation，所以DispatchedContinuation繼承的父類中，必定有繼承了Runnable接口，而他的run方法的實現也在父類中，這塊我們暫時按下不表，接著看繼續跟蹤coroutineScheduler.dispatch(block)。

1.5.2 CoroutineScheduler

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
        val task = createTask(block, taskContext)
        // try to submit the task to the local queue and act depending on the result
        val currentWorker = currentWorker()
        val notAdded = currentWorker.submitToLocalQueue(task, tailDispatch)
        if (notAdded != null) {
            if (!addToGlobalQueue(notAdded)) {
                // Global queue is closed in the last step of close/shutdown -- no more tasks should be accepted
                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
            signalCpuWork()
        } else {
            // Increment blocking tasks anyway
            signalBlockingWork(skipUnpark = skipUnpark)
        }
    }
}

該類繼承了Executor類，而且它的構建參數可看到是線程池的參數，所以可以知道這個其實是Kotlin協程實現的一個線程池，具體就不跟進去了。
execute()過程也是dispatch過程：將任務投遞到任務隊列，然后通知線程去取任務執行，自此完成了線程切換動作。
而在新線程里執行的Runnable為1.4中的調用代碼：dispatcher.dispatch(context, this)中的this，也就是DispatchedContinuation。DispatchedContinuation.kt并沒有實現run方法，那么一定是他繼承的父類實現了Runnable接口并實現，所以需要接著看它繼承的父類：DispatchedTask類。

1.6 DispatchedTask.run()

internal abstract class DispatchedTask<in T>(
    @JvmField public var resumeMode: Int
) : SchedulerTask() {
	...
    internal abstract val delegate: Continuation<T>
    @Suppress("UNCHECKED_CAST")
    internal open fun <T> getSuccessfulResult(state: Any?): T =
        state as T
    internal open fun getExceptionalResult(state: Any?): Throwable? =
        (state as? CompletedExceptionally)?.cause
    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>
            //1. 取出代理商的續體
            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)
                val job = if (exception == null && resumeMode.isCancellableMode) context[Job] else null
                if (job != null && !job.isActive) {
                    val cause = job.getCancellationException()
                    cancelCompletedResult(state, cause)
                    continuation.resumeWithStackTrace(cause)
                } else {
                    if (exception != null) {
                        continuation.resumeWithException(exception)
                    } else {
                        //1. 被包裝的續體的resume方法，真正的開始出發其協程狀態機代碼。
                        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())
        }
    }
}

將delegate轉為DispatchedContinuation，應該注意1.4 小節中DispatchedContinuation繼承DispatchTask時，便對此delegate進行了復寫：

override val delegate: Continuation

get() = this

而此delegate.continucation便是當初newDispatchedContinuation(this)時傳入的this，此this就是Kotlin編譯器一開始為協程體生成的SuspendLambda類型對象。具體可以回看1.3小節。

調用了continuation.resume()方法觸發了協程的狀態機進而開始執行協程業務邏輯代碼，結合之前1.5.2的分析可以知道，這個方法的調用已經是被dispatch到特定線程，完成線程切換后執行的。所以協程狀態機的代碼也是跑在新線程上的。

1.7 總結

至此，協程的線程調度分析結束，關鍵有如下幾個要點：

創建SuspendLambda時，他的協程上下文對象來自于comletion.context，默認就是Dispatcher.Default。
SuspendLambda啟動時調用了intercept()進行一層包裝，得到DispatchedContinuation，后續協程啟動是啟動的DispatchedContinuation協程。
DispatchedContinuation繼承于Runnable接口，協程啟動時將自己投遞到分發器dispatcher執行run方法，從而達到了線程切換效果。
在DispatchedContinuation的run方法中，調用SuspendLambda.resume()啟動狀態機。在新線程執行協程狀態機代碼。

這一小節中，介紹了如何將協程調度到目的線程執行，接下來分析如何做到隨意切換線程后，然后再恢復到原來線程的。

二、協程中的線程切換

在第一小節中，我們搞清楚了協程啟動時，協程調度器是如何在其中起作用的。這一小節旨在剖析在協程用分發器切換線程執行新的掛起函數后，是如何切換會原來線程繼續執行剩下的邏輯的。

為此，我們需要將1.1的測試代碼反編譯出來實際代碼進而分析。

2.1 反編譯代碼

2.1.1 MainActivityonCreateonCreateonCreate1

final class MainActivity$onCreate$1 extends SuspendLambda implements Function2<CoroutineScope, Continuation<? super Unit>, Object> {
    ...
    @Override // kotlin.coroutines.jvm.internal.BaseContinuationImpl
    public final Object invokeSuspend(Object $result) {
        Object coroutine_suspended = IntrinsicsKt.getCOROUTINE_SUSPENDED();
        switch (this.label) {
            case 0:
                ResultKt.throwOnFailure($result);
                Log.d(MainActivity.TAG, LiveLiterals$MainActivityKt.INSTANCE.m4147xf96cab04());
                this.label = 1;
                //1. 新建編譯器自動生成的繼承于SuspendLambda的類型。
                AnonymousClass1 anonymousClass1 = new AnonymousClass1(null);
                //2. 調用withContext
            	Object res = BuildersKt.withContext(Dispatchers.getIO(), anonymousClass1, this);
                if (res != coroutine_suspended) {
                    break;
                } else {
                    //掛起
                    return coroutine_suspended;
                }
            case 1:
                ResultKt.throwOnFailure($result);
                break;
            default:
                throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
        }
        Log.d(MainActivity.TAG, LiveLiterals$MainActivityKt.INSTANCE.m4148xe0c1b328());
        return Unit.INSTANCE;
    }
}

根據之前的文章分析，這里suspend lambda 的類型都自動生成繼承于SuspendLambda的類型。詳見2.1.2。

將anonymousClass1傳入withContext，而且注意這里傳入了this==MainActivity$onCreate$1，詳見2.2。

2.1.2 AnonymousClass1

/* compiled from: MainActivity.kt */
public static final class AnonymousClass1 extends SuspendLambda implements Function2<CoroutineScope, Continuation<? super Integer>, Object> {
    int label
    ...
    @Override // kotlin.coroutines.jvm.internal.BaseContinuationImpl
    public final Object invokeSuspend(Object obj) {
        IntrinsicsKt.getCOROUTINE_SUSPENDED();
        switch (this.label) {
            case 0:
                ResultKt.throwOnFailure(obj);
                return Boxing.boxInt(Log.d(MainActivity.TAG, LiveLiterals$MainActivityKt.INSTANCE.m4146x7c0f011f()));
            default:
                throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
        }
    }
}

2.2 withContext

public suspend fun <T> withContext(
    context: CoroutineContext,
    block: suspend CoroutineScope.() -> T
): T {
    contract {
        callsInPlace(block, InvocationKind.EXACTLY_ONCE)
    }
    //1. 獲取當前協程, 注意這里的uCont就是當前續體，也就是MainActivity$onCreate$1
    return suspendCoroutineUninterceptedOrReturn sc@ { uCont ->
        //2. 計算獲的新的協程上下文
        val oldContext = uCont.context
        val newContext = oldContext + context
        //3. 快速判斷：新上下文和舊上下文一致的情況快速處理。
        // always check for cancellation of new context
        newContext.ensureActive()
        // FAST PATH #1 -- new context is the same as the old one
        if (newContext === oldContext) {
            val coroutine = ScopeCoroutine(newContext, uCont)
            return@sc coroutine.startUndispatchedOrReturn(coroutine, block)
        }
        // FAST PATH #2 -- the new dispatcher is the same as the old one (something else changed)
        // `equals` is used by design (see equals implementation is wrapper context like ExecutorCoroutineDispatcher)
        if (newContext[ContinuationInterceptor] == oldContext[ContinuationInterceptor]) {
            val coroutine = UndispatchedCoroutine(newContext, uCont)
            // There are changes in the context, so this thread needs to be updated
            withCoroutineContext(newContext, null) {
                return@sc coroutine.startUndispatchedOrReturn(coroutine, block)
            }
        }
        // SLOW PATH -- use new dispatcher
        //4. 新建一個DispatchedCoroutine
        val coroutine = DispatchedCoroutine(newContext, uCont)
        //5. 啟動協程
        block.startCoroutineCancellable(coroutine, coroutine)
        coroutine.getResult()
    }
}

suspendCoroutineUninterceptedOrReturn這個函數直接步進是看不到實現的，它的實現是由Kotlin編譯器生成的，它的作用是用來獲取當前續體的，并且通過uCont返回，這里就是MainActivity$onCreate$1。
將舊協程上下文和新的上下文一起。計算得到最終的上下文。這里的context==Dispatchers.getIO()。
快速判斷，不用看。
新建一個DispatchedCoroutine，注意這里傳入了新的協程上下文和當前續體對象。
調用startCoroutineCancellable()啟動協程。這里的同1.3.2小節分析一樣，詳見 2.2.1

2.2.1 startCoroutineCancellable

internal fun <R, T> (suspend (R) -> T).startCoroutineCancellable(
    receiver: R, completion: Continuation<T>,
    onCancellation: ((cause: Throwable) -> Unit)? = null
) =
    runSafely(completion) {
    	//1. 創建SuspendLambda協程體
        createCoroutineUnintercepted(receiver, completion)
            //2. 攔截：取出分發器，并構建方法器Continuation。詳見1.3.1
            .intercepted()
            //3. 調用方法器Continuation的resume方法，詳見1.4
            .resumeCancellableWith(Result.success(Unit), onCancellation)
    }

此方法在之前1.3小節已經分析過，針對此此次調用，其中的改變是協程上下文中的分發器已經被設置為Dispatchers.Main。

創建了SuspendLambda對象，此對象的CoroutineContext為completion.context。而其中的ContinuationInterceptor類型Element就是我們之前傳入的Dispatchers.Main。
創建一個DispatchedContinuation。
將協程SuspendLambda的狀態機邏輯通過Dispatcher.Main調度到主線程執行，調度過程參考第一下節。分發邏輯詳見2.7小節。
當SuspendLambda的狀態機invokeSuspend()邏輯執行完成后，會返回到BaseContinuationImpl.resumeWith()，我們需要接此方法分析，來得到協程在切換到主線程執行后，又是怎么切回協程體1的執行線程的，詳見2.3。

2.3 resumeWith

public final override fun resumeWith(result: Result<Any?>) {
        // This loop unrolls recursion in current.resumeWith(param) to make saner and shorter stack traces on resume
        var current = this
        var param = result
        while (true) {
            // Invoke "resume" debug probe on every resumed continuation, so that a debugging library infrastructure
            // can precisely track what part of suspended callstack was already resumed
            probeCoroutineResumed(current)
            with(current) {
                val completion = completion!! // fail fast when trying to resume continuation without completion
                val outcome: Result<Any?> =
                    try {
                        val outcome = invokeSuspend(param)
                        if (outcome === COROUTINE_SUSPENDED) return
                        Result.success(outcome)
                    } catch (exception: Throwable) {
                        Result.failure(exception)
                    }
                releaseIntercepted() // this state machine instance is terminating
                if (completion is BaseContinuationImpl) {
                    // unrolling recursion via loop
                    current = completion
                    param = outcome
                } else {
                    //1. 進入此判斷
                    // top-level completion reached -- invoke and return
                    completion.resumeWith(outcome)
                    return
                }
            }
        }
    }

當狀態機執行完后，后進入到completion的類型判斷，由2.2和2.2.1可以知道，當初傳入的completion是DispatchedCoroutine類型，所以加入到else分支，調用了DispatchedCoroutine.resumeWith()，接下來分析此方法。

在此之前，我們需要看下DispatchedCoroutine的繼承關系，詳見2.4.1。如果想直接跟蹤流程，可以直接看2.4.2。

2.4 DispatchedCoroutine

2.4.1 DispatchedCoroutine 的繼承關系

internal class DispatchedCoroutine<in T>(
    context: CoroutineContext,
    uCont: Continuation<T>
) : ScopeCoroutine<T>(context, uCont) {
}

繼承于ScopeCoroutine

internal open class ScopeCoroutine<in T>(
    context: CoroutineContext,
    @JvmField val uCont: Continuation<T> // unintercepted continuation
) : AbstractCoroutine<T>(context, true, true), CoroutineStackFrame {
}

繼承于AbstractCoroutine

public abstract class AbstractCoroutine<in T>(
    parentContext: CoroutineContext,
    initParentJob: Boolean,
    active: Boolean
) : JobSupport(active), Job, Continuation<T>, CoroutineScope {
}

2.5 協程線程的恢復

2.5.1 AbstractCoroutine.resumeWith()

    public final override fun resumeWith(result: Result<T>) {
        val state = makeCompletingOnce(result.toState())
        if (state === COMPLETING_WAITING_CHILDREN) return
        afterResume(state)
    }

調用了afterResume方法，此方法在DispatchedCoroutine類型有具體實現。見2.5.2

2.5.2 afterResume

//DispatchedCoroutine
override fun afterResume(state: Any?) {
        if (tryResume()) return // completed before getResult invocation -- bail out
        // Resume in a cancellable way because we have to switch back to the original dispatcher
        uCont.intercepted().resumeCancellableWith(recoverResult(state, uCont))
}

取出當前續體uCont，這個續體根據之前的分析：2.2小節，可以知道它等于MainActivity$onCreate$1。
intercepted()：取出其分發攔截器
resumeCancellableWith：使用方法攔截器協程體，將uCont續體的狀態機邏輯調度到相對應的線程環境執行，這里就是之前的Dispatcher.Default。注意其注釋：“將其切換到原先的分發器”。2?而這一過程其實和1.3小節的過程一致。
恢復到Dispatcher.Default繼續執行狀態機時，由于label已經被更新，所以會往下繼續執行，打印最后一句log。

2.6 總結

withContext(Dispatcher.Main)啟動的協程時，取得當前協程續體uCount也就是MainActivity$onCreate$1，會計算出新的協程context，然后用它們創建一個DispatchedCoroutine。

AnonymousClass1協程啟動時，用DispatchedCoroutine作為completion參數，然后啟動，此時會調度主線程執行協程。

當協程執行完成后，AnonymousClass1.resumeWith()方法會調用completion.resumeWith()。

DispatchedCoroutine.resumeWith()方法會調用uCount.intercepted().resumeCancellableWith()，使得父協程進行調度并接著執行狀態機邏輯。

2.7 Dispatchers.Main

    @JvmStatic
    public actual val Main: MainCoroutineDispatcher get() 
= MainDispatcherLoader.dispatcher

直接詳見2.7.1

2.7.1 MainDispatcherLoader

internal object MainDispatcherLoader {
    private val FAST_SERVICE_LOADER_ENABLED = systemProp(FAST_SERVICE_LOADER_PROPERTY_NAME, true)
    @JvmField
    val dispatcher: MainCoroutineDispatcher = loadMainDispatcher()
    private fun loadMainDispatcher(): MainCoroutineDispatcher {
        return try {
            val factories = if (FAST_SERVICE_LOADER_ENABLED) {
                FastServiceLoader.loadMainDispatcherFactory()
            } else {
                // We are explicitly using the
                // `ServiceLoader.load(MyClass::class.java, MyClass::class.java.classLoader).iterator()`
                // form of the ServiceLoader call to enable R8 optimization when compiled on Android.
                // 1.獲得MainDispatcherFactory的實現類
                ServiceLoader.load(
                        MainDispatcherFactory::class.java,
                        MainDispatcherFactory::class.java.classLoader
                ).iterator().asSequence().toList()
            }
            @Suppress("ConstantConditionIf")
            factories.maxByOrNull { it.loadPriority }?.tryCreateDispatcher(factories)
                ?: createMissingDispatcher()
        } catch (e: Throwable) {
            // Service loader can throw an exception as well
            createMissingDispatcher(e)
        }
    }
}

通過ServiceLoad機制獲取MainDispatcherFactory的實現類，而在源碼里面，其實現類為AndroidDispatcherFactory
調用tryCreateDispatcher()創建分發器，詳見2.7.2。

2.7.2 AndroidDispatcherFactory

internal class AndroidDispatcherFactory : MainDispatcherFactory {
    override fun createDispatcher(allFactories: List<MainDispatcherFactory>) =
        HandlerContext(Looper.getMainLooper().asHandler(async = true))
    override fun hintOnError(): String = "For tests Dispatchers.setMain from kotlinx-coroutines-test module can be used"
    override val loadPriority: Int
        get() = Int.MAX_VALUE / 2
}

根據createDispatcher分發，主線程分發器的實現類為HandlerContext類型，傳入用MainLooper構建的Handler。詳見2.7.3。

2.7.3 HandlerContext

internal class HandlerContext private constructor(
    private val handler: Handler,
    private val name: String?,
    private val invokeImmediately: Boolean
) : HandlerDispatcher(), Delay {
    /**
     * Creates [CoroutineDispatcher] for the given Android [handler].
     *
     * @param handler a handler.
     * @param name an optional name for debugging.
     */
    constructor(
        handler: Handler,
        name: String? = null
    ) : this(handler, name, false)
    @Volatile
    private var _immediate: HandlerContext? = if (invokeImmediately) this else null
    override val immediate: HandlerContext = _immediate ?:
        HandlerContext(handler, name, true).also { _immediate = it }
    override fun isDispatchNeeded(context: CoroutineContext): Boolean {
        return !invokeImmediately || Looper.myLooper() != handler.looper
    }
    override fun dispatch(context: CoroutineContext, block: Runnable) {
        if (!handler.post(block)) {
            cancelOnRejection(context, block)
        }
    }
    override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
        val block = Runnable {
            with(continuation) { resumeUndispatched(Unit) }
        }
        if (handler.postDelayed(block, timeMillis.coerceAtMost(MAX_DELAY))) {
            continuation.invokeOnCancellation { handler.removeCallbacks(block) }
        } else {
            cancelOnRejection(continuation.context, block)
        }
    }
   ...
}

HandlerContext繼承于HandlerDispatcher，而他的dispatch方法，可以看到，就是將block丟到設置MainLooper的handler執行。所以續體將會在主線程執行狀態機，達到切換到主線程執行協程的目的。

原文鏈接：https://juejin.cn/post/7175783679993020453

上一篇：No?module?named?'plotly.graph_
下一篇：Pyscript使用本地Pyodide配置步驟_python

日本免费高清视频-国产福利视频导航-黄色在线播放国产-天天操天天操天天操天天操|www.shdianci.com

網站首頁編程語言正文

Kotlin協程的線程調度示例詳解_Android

目錄

引言

一、協程的分發器作用

1.1 測試代碼

1.2 CoroutineScope.launch

1.2.1 newCoroutineContext

1.3 startCoroutineCancellable

1.3.1 intercepted()

1.3.2 CoroutineDispatcher

1.3.3 小結

1.4 DispatchedContinuation

1.5 DefaultScheduler

1.5.1 SchedulerCoroutineDispatcher

1.5.2 CoroutineScheduler

1.6 DispatchedTask.run()

1.7 總結

二、協程中的線程切換

2.1 反編譯代碼

2.1.1 MainActivityonCreateonCreateonCreate1

2.1.2 AnonymousClass1

2.2 withContext

2.2.1 startCoroutineCancellable

2.3 resumeWith

2.4 DispatchedCoroutine

2.4.1 DispatchedCoroutine 的繼承關系

2.5 協程線程的恢復

2.5.1 AbstractCoroutine.resumeWith()

2.5.2 afterResume

2.6 總結

2.7 Dispatchers.Main

2.7.1 MainDispatcherLoader

2.7.2 AndroidDispatcherFactory

2.7.3 HandlerContext

相關推薦

日本免费高清视频-国产福利视频导航-黄色在线播放国产-天天操天天操天天操天天操|www.shdianci.com

網站首頁 編程語言 正文

Kotlin協程的線程調度示例詳解_Android

目錄

引言

一、協程的分發器作用

1.1 測試代碼

1.2 CoroutineScope.launch

1.2.1 newCoroutineContext

1.3 startCoroutineCancellable

1.3.1 intercepted()

1.3.2 CoroutineDispatcher

1.3.3 小結

1.4 DispatchedContinuation

1.5 DefaultScheduler

1.5.1 SchedulerCoroutineDispatcher

1.5.2 CoroutineScheduler

1.6 DispatchedTask.run()

1.7 總結

二、協程中的線程切換

2.1 反編譯代碼

2.1.1 MainActivityonCreateonCreateonCreate1

2.1.2 AnonymousClass1

2.2 withContext

2.2.1 startCoroutineCancellable

2.3 resumeWith

2.4 DispatchedCoroutine

2.4.1 DispatchedCoroutine 的繼承關系

2.5 協程線程的恢復

2.5.1 AbstractCoroutine.resumeWith()

2.5.2 afterResume

2.6 總結

2.7 Dispatchers.Main

2.7.1 MainDispatcherLoader

2.7.2 AndroidDispatcherFactory

2.7.3 HandlerContext

相關推薦

網站首頁編程語言正文