Kotlin线程的桥接与切换使用介绍

@Throws(InterruptedException::class)
public fun  runBlocking(context: CoroutineContext = EmptyCoroutineContext, block: suspend CoroutineScope.() -> T): T {
    // 通知编译器，block只执行一次
    contract {
        callsInPlace(block, InvocationKind.EXACTLY_ONCE)
    }
    // 获取当前线程
    val currentThread = Thread.currentThread()
    // 获取上下文中的拦截器
    val contextInterceptor = context[ContinuationInterceptor]
    val eventLoop: EventLoop?
    val newContext: CoroutineContext
    // 如果拦截器为空，代表无法进行调度
    if (contextInterceptor == null) {
        // 从线程中获取EventLoop，获取失败则创建一个新的
        eventLoop = ThreadLocalEventLoop.eventLoop
        // 添加到上下文中
        // newContext = EmptyCoroutineContext + context + eventLoop
        newContext = GlobalScope.newCoroutineContext(context + eventLoop)
    } else {// 如果有拦截器
        // 尝试将当前拦截器转换成EventLoop，
        // 如果转换成功，则判断是否允许可以在上下文中使用
        // 如果转换失败或不允许，则创建一个新的
        eventLoop = (contextInterceptor as? EventLoop)?.takeIf { it.shouldBeProcessedFromContext() }
            ?: ThreadLocalEventLoop.currentOrNull()
        // 计算新的上下文
        // 这里没有把EventLoop加到上下文，因为加入后会覆盖拦截器
        newContext = GlobalScope.newCoroutineContext(context)
    }
    // 创建一个协程
    val coroutine = BlockingCoroutine(newContext, currentThread, eventLoop)
    // 启动协程
    coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
    // 分发任务
    return coroutine.joinBlocking()
}

// 继承了AbstractCoroutine
private class BlockingCoroutine(
    parentContext: CoroutineContext,
    private val blockedThread: Thread,
    private val eventLoop: EventLoop?
) : AbstractCoroutine(parentContext, true) {
    // 该协程是一个作用域协程
    override val isScopedCoroutine: Boolean get() = true

    override fun afterCompletion(state: Any?) {
        // 如果当前线程不是阻塞线程
        if (Thread.currentThread() != blockedThread)
            // 唤醒阻塞线程
            LockSupport.unpark(blockedThread)
    }

    @Suppress("UNCHECKED_CAST")
    fun joinBlocking(): T {
        registerTimeLoopThread()
        try {
            // 注册使用EventLoop
            eventLoop?.incrementUseCount()
            try {
                // 死循环
                while (true) {
                    @Suppress("DEPRECATION")
                    // 如果线程当前中断，则抛出异常，同时取消当前协程
                    if (Thread.interrupted()) throw InterruptedException().also { cancelCoroutine(it) }
                    // 分发执行任务，同时获取等待时间
                    val parkNanos = eventLoop?.processNextEvent() ?: Long.MAX_VALUE
                    // 如果任务执行结束，则退出循环
                    if (isCompleted) break
                    // 休眠指定的等待时间
                    parkNanos(this, parkNanos)
                }
            } finally { // paranoia
                // 注册不使用EventLoop
                eventLoop?.decrementUseCount()
            }
        } finally { // paranoia
            unregisterTimeLoopThread()
        }
        // 获取执行的结果
        val state = this.state.unboxState()
        // 如果执行过程中取消，则抛出异常
        (state as? CompletedExceptionally)?.let { throw it.cause }
        // 返回结果
        return state as T
    }
}

public suspend fun  withContext(
    context: CoroutineContext,
    block: suspend CoroutineScope.() -> T
): T {
    // 通知编译器，block只执行一次
    contract {
        callsInPlace(block, InvocationKind.EXACTLY_ONCE)
    }
    // 直接挂起，获取续体
    return suspendCoroutineUninterceptedOrReturn sc@ { uCont ->
        // 从续体中获取上下文
        val oldContext = uCont.context
        // 计算新的上下文
        val newContext = oldContext + context
        // 检查任务是否执行完毕或取消
        newContext.checkCompletion()
        // 如果前后两次的上下文完全相同，说明不需要切换，只需要执行即可
        if (newContext === oldContext) {
            // 创建续体的协程
            val coroutine = ScopeCoroutine(newContext, uCont)
            // 执行block
            return@sc coroutine.startUndispatchedOrReturn(coroutine, block)
        }
        // 拦截器相同，但是上下文中增加了其他的元素
        // 这里也是在同一个线程上执行，但是其中增加的元素只在执行当前的block中使用
        if (newContext[ContinuationInterceptor] == oldContext[ContinuationInterceptor]) {
            // 创建续体的协程
            val coroutine = UndispatchedCoroutine(newContext, uCont)
            // 将当前线程ThreadLocal中的对象更新成newContext上下文对应的对象
            withCoroutineContext(newContext, null) {
                // 执行block
                return@sc coroutine.startUndispatchedOrReturn(coroutine, block)
            }
        }
        // 走到这里，说明要切换线程执行block任务
        val coroutine = DispatchedCoroutine(newContext, uCont)
        // 启动父协程
        coroutine.initParentJob()
        // 启动协程
        block.startCoroutineCancellable(coroutine, coroutine)
        // 获取结果
        coroutine.getResult()
    }
}

internal fun  ScopeCoroutine.startUndispatchedOrReturn(receiver: R, block: suspend R.() -> T): Any? {
    // 初始化并绑定父协程
    initParentJob()
    // 获取并处理执行结果
    return undispatchedResult({ true }) {
        // 启动协程
        block.startCoroutineUninterceptedOrReturn(receiver, this)
    }
}
private inline fun  ScopeCoroutine.undispatchedResult(
    shouldThrow: (Throwable) -> Boolean,
    startBlock: () -> Any?
): Any? {
    // 启动协程，获取结果，
    val result = try {
        startBlock()
    } catch (e: Throwable) {
        // 产生异常，则按照取消处理
        CompletedExceptionally(e)
    }
    // 如果结果为挂起，则通知外部挂起    
    if (result === COROUTINE_SUSPENDED) return COROUTINE_SUSPENDED
    // 结束任务执行，获取最终状态
    val state = makeCompletingOnce(result)
    // 如果需要等待子协程的结束，则通知外部挂起
    if (state === COMPLETING_WAITING_CHILDREN) return COROUTINE_SUSPENDED
    // 如果执最终为异常状态
    return if (state is CompletedExceptionally) {
        when {
            // 通过参数判断是否抛出
            shouldThrow(state.cause) -> throw recoverStackTrace(state.cause, uCont)
            // 执行结果为异常
            result is CompletedExceptionally -> throw recoverStackTrace(result.cause, uCont)
            // 结果不为异常，则返回
            else -> result
        }
    } else {
        // 对最终状态进行拆箱，返回最终结果
        state.unboxState()
    }
}
// JobSupport中提供了下面的类和方法，当协程进入完成状态时，会对状态进行装箱。
// 包装类
private class IncompleteStateBox(@JvmField val state: Incomplete)
// 装箱
internal fun Any?.boxIncomplete(): Any? = if (this is Incomplete) IncompleteStateBox(this) else this
// 拆箱
internal fun Any?.unboxState(): Any? = (this as? IncompleteStateBox)?.state ?: this

@SinceKotlin("1.3")
@InlineOnly
public actual inline fun  (suspend R.() -> T).startCoroutineUninterceptedOrReturn(
    receiver: R,
    completion: Continuation
): Any? = (this as Function2, Any?>).invoke(receiver, completion)

internal open class ScopeCoroutine(
    context: CoroutineContext,
    @JvmField val uCont: Continuation
) : AbstractCoroutine(context, true), CoroutineStackFrame {
    final override val callerFrame: CoroutineStackFrame? get() = uCont as CoroutineStackFrame?
    final override fun getStackTraceElement(): StackTraceElement? = null
    // 作用域协程
    final override val isScopedCoroutine: Boolean get() = true
    internal val parent: Job? get() = parentContext[Job]
    // 该方法会在协程异常或取消时调用
    override fun afterCompletion(state: Any?) {
        // 进行拦截，切换线程，恢复执行
        uCont.intercepted().resumeCancellableWith(recoverResult(state, uCont))
    }
    // 该方法会在将其挂起的方法执行完毕后回调
    override fun afterResume(state: Any?) {
        // 直接恢复续体的执行
        uCont.resumeWith(recoverResult(state, uCont))
    }
}

private class UndispatchedCoroutine(
    context: CoroutineContext,
    uCont: Continuation
) : ScopeCoroutine(context, uCont) {
    override fun afterResume(state: Any?) {
        val result = recoverResult(state, uCont)
        // 将当前线程ThreadLocal中的对象更新成uCont.context上下文对应的对象
        withCoroutineContext(uCont.context, null) {
            // 恢复执行
            uCont.resumeWith(result)
        }
    }
}

internal actual inline fun  withCoroutineContext(context: CoroutineContext, countOrElement: Any?, block: () -> T): T {
    // 将线程上下文更新新的上下文，并返回老的上下文
    val oldValue = updateThreadContext(context, countOrElement)
    try {
        // 在新的上下文环境中执行
        return block()
    } finally {
        // 执行结束恢复老的上下文
        restoreThreadContext(context, oldValue)
    }
}

public interface ThreadContextElement : CoroutineContext.Element {
    // 用于更新新的上下文，并且返回老的上下文
    public fun updateThreadContext(context: CoroutineContext): S
    // 重新恢复当前线程的上下文，
    // 其中oldStart来自updateThreadContext方法的返回值
    public fun restoreThreadContext(context: CoroutineContext, oldState: S)
}