目录
- 一.线程的桥接
- 1.runBlocking方法
- 2.BlockingCoroutine类
- 二.线程的切换
- 1.withContext方法
- 2.startUndispatchedOrReturn方法
- 3.ScopeCoroutine类
- 4.UndispatchedCoroutine类
- 5.DispatchedCoroutine类
- 6.总结
一.线程的桥接
1.runBlocking方法
runBlocking方法用于在线程中去执行suspend方法，代码如下：
@Throws(InterruptedException::class)
public fun <T> 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<T>(newContext, currentThread, eventLoop)
// 启动协程
coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
// 分发任务
return coroutine.joinBlocking()
}
2.BlockingCoroutine类
在runBlocking方法中，最终创建了一个类型为BlockingCoroutine的对象。BlockingCoroutine类继承自AbstractCoroutine类，代码如下：
// 继承了AbstractCoroutine
private class BlockingCoroutine<T>(
parentContext: CoroutineContext,
private val blockedThread: Thread,
private val eventLoop: EventLoop?
) : AbstractCoroutine<T>(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
}
}
BlockingCoroutine类重写了变量isScopedCoroutine为true。
isScopedCoroutine表示当前协程是否为作用域协程，该变量用在cancelParent方法中。对于一个作用域协程，当它的子协程在运行过程中抛出异常时，子协程调用cancelParent方法不会导致作用域协程取消，而是直接返回true。当子协程执行完毕，作用域协程获取结果时，如果发现子协程返回的结果为异常，则会再次抛出。
相比于一般协程，作用域协程不相信子协程在执行过程中取消通知，而是在执行完毕后亲自检查结果是否为异常，达到一种“耳听为虚，眼见为实”的效果。
joinBlocking方法通过循环在当前线程上对EventLoop进行任务分发来实现线程的阻塞。当任务发生异常或执行完毕后，会回调重写的afterCompletion方法，唤起线程继续循环，当在循环中检测到isCompleted标志位为true时，会跳出循环，恢复线程执行。
二.线程的切换
1.withContext方法
withContext方法用于在协程中切换线程去执行其他任务，该方法被suspend关键字修饰，因此会引起协程的挂起，代码如下：
public suspend fun <T> 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()
}
}
通过对上面代码的分析，可以发现withContext根据上下文的不同进行了三种分类，创建不同的协程并通过不同的方式去执行block。如下表所示：
接下来，将对不同情况下协程的启动与执行进行分析。
2.startUndispatchedOrReturn方法
startUndispatchedOrReturn方法用于在相同的上下文环境中启动协程，代码如下：
internal fun <T, R> ScopeCoroutine<T>.startUndispatchedOrReturn(receiver: R, block: suspend R.() -> T): Any? {
// 初始化并绑定父协程
initParentJob()
// 获取并处理执行结果
return undispatchedResult({ true }) {
// 启动协程
block.startCoroutineUninterceptedOrReturn(receiver, this)
}
}
private inline fun <T> ScopeCoroutine<T>.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
在startUndispatchedOrReturn方法中，通过调用block的startCoroutineUninterceptedOrReturn方法启动协程，获取最终结果，并对结果进行异常处理。
接下来，将分析startCoroutineUninterceptedOrReturn方法如何启动协程，代码如下：
@SinceKotlin("1.3")
@InlineOnly
public actual inline fun <R, T> (suspend R.() -> T).startCoroutineUninterceptedOrReturn(
receiver: R,
completion: Continuation<T>
): Any? = (this as Function2<R, Continuation<T>, Any?>).invoke(receiver, completion)
这里，直接找到最终的actual方法，可以发现该方法没有创建状态机，而是直接执行了block。这个方法被设计用在suspendCoroutineUninterceptedOrReturn方法中，来恢复挂起协程的执行。
至此，可以知道startUndispatchedOrReturn方法实际上就是在同一个协程中执行了block。
3.ScopeCoroutine类
在withContext方法中，当上下文相同时，会创建一个类型为ScopeCoroutine的对象。ScopeCoroutine类代表一个标准的作用域协程，代码如下：
internal open class ScopeCoroutine<in T>(
context: CoroutineContext,
@JvmField val uCont: Continuation<T>
) : AbstractCoroutine<T>(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))
}
}
ScopeCoroutine类重写了afterCompletion和afterResume两个方法，afterCompletion方法用于在协程取消时被回调。afterResume方法用于在挂起恢复时被回调。
根据上面的分析，当发生异常时，afterCompletion方法可能在其他的协程上下文中被调用，因此会调用拦截器切换回原本的线程中。而afterResume方法由于已经在正确的上下文环境中，因此可以直接恢复执行。
4.UndispatchedCoroutine类
在withContext方法中，当上下文不同，但调度器相同时，会创建一个类型为UndispatchedCoroutine的对象。UndispatchedCoroutine类继承自ScopeCoroutine类，重写了afterResume方法，代码如下：
private class UndispatchedCoroutine<in T>(
context: CoroutineContext,
uCont: Continuation<T>
) : ScopeCoroutine<T>(context, uCont) {
override fun afterResume(state: Any?) {
val result = recoverResult(state, uCont)
// 将当前线程ThreadLocal中的对象更新成uCont.context上下文对应的对象
withCoroutineContext(uCont.context, null) {
// 恢复执行
uCont.resumeWith(result)
}
}
}
与父类ScopeCoroutine的afterResume方法相比，UndispatchedCoroutine类在afterResume方法中对协程上下文进行了更新，然后再恢复执行。
- withCoroutineContext
withCoroutineContext方法用于当一个线程中执行多个协程时，保存和恢复ThreadLocal类中的对象。
通过withContext方法的代码可以知道，当上下文不同但调度器相同时，在执行之前会通过withCoroutineContext方法将ThreadLocal中的对象更新成newContext对应的对象。在执行结束后，又将ThradLocal中的对象更新成原本续体的上下文context对应的对象。代码如下：
internal actual inline fun <T> withCoroutineContext(context: CoroutineContext, countOrElement: Any?, block: () -> T): T {
// 将线程上下文更新新的上下文，并返回老的上下文
val oldValue = updateThreadContext(context, countOrElement)
try {
// 在新的上下文环境中执行
return block()
} finally {
// 执行结束恢复老的上下文
restoreThreadContext(context, oldValue)
}
}
协程中有一类上下文元素是ThreadContextElement，ThreadContextElement是一个接口，具体的实现类有CoroutineId类和ThreadLocalElement类。其中，CoroutineId类用来修改线程的名字。ThreadLocalElement类用来保存和恢复ThreadLocal类中的对象，withCoroutineContext方法内部的updateThreadContext方法与restoreThreadContext方法正是通过ThreadLocalElement类实现的。ThreadContextElement接口的代码如下：
public interface ThreadContextElement<S> : CoroutineContext.Element {
// 用于更新新的上下文，并且返回老的上下文
public fun updateThreadContext(context: CoroutineContext): S
// 重新恢复当前线程的上下文，
// 其中oldStart来自updateThreadContext方法的返回值
public fun restoreThreadContext(context: CoroutineContext, oldState: S)
}
当调用updateThreadContext方法时，会返回一个代表当前状态的对象。当调用restoreThreadContext方法时，又需要传入一个代表状态的对象作为参数，来恢复之前的状态。因此，这就需要对updateThreadContext方法的返回值进行保存。
当协程上下文中只有一个ThreadContextElement接口指向的对象时，保存在变量中即可。而如果协程上下文中有多个ThreadContextElement接口指向的对象，这时就需要一个专门的类来对这些对象进行管理，这个类就是ThreadState类，他们之间的对应关系如下图所示：
withCoroutineContext方法执行图：
5.DispatchedCoroutine类
在withContext方法中，当需要切换线程时，会创建一个类型为DispatchedCoroutine的对象。DispatchedCoroutine类继承自ScopeCoroutine类，代码如下：
// 状态机状态
private const val UNDECIDED = 0
private const val SUSPENDED = 1
private const val RESUMED = 2
private class DispatchedCoroutine<in T>(
context: CoroutineContext,
uCont: Continuation<T>
) : ScopeCoroutine<T>(context, uCont) {
// 初始状态
private val _decision = atomic(UNDECIDED)
// 尝试挂起
private fun trySuspend(): Boolean {
_decision.loop { decision ->
when (decision) {
UNDECIDED -> if (this._decision.compareAndSet(UNDECIDED, SUSPENDED)) return true
RESUMED -> return false
else -> error("Already suspended")
}
}
}
// 尝试恢复
private fun tryResume(): Boolean {
_decision.loop { decision ->
when (decision) {
UNDECIDED -> if (this._decision.compareAndSet(UNDECIDED, RESUMED)) return true
SUSPENDED -> return false
else -> error("Already resumed")
}
}
}
override fun afterCompletion(state: Any?) {
// 通过afterResume方法实现
afterResume(state)
}
override fun afterResume(state: Any?) {
// 如果没有挂起，则返回
if (tryResume()) return
// 进行拦截，切换线程，恢复执行
uCont.intercepted().resumeCancellableWith(recoverResult(state, uCont))
}
// 获取最终结果
fun getResult(): Any? {
if (trySuspend()) return COROUTINE_SUSPENDED
val state = this.state.unboxState()
if (state is CompletedExceptionally) throw state.cause
@Suppress("UNCHECKED_CAST")
return state as T
}
}
DispatchedCoroutine类中使用了一个状态机模型，这个状态机与在Kotlin协程：生命周期原理中分析CancellableContinuationImpl类中的状态机相同，获取结果的逻辑也与CancellableContinuationImpl类相同。
这里最重要的是DispatchedCoroutine类重写了afterCompletion和afterResume方法，并且回调这两个方法都会进行线程的切换。
6.总结
以上就是Kotlin线程的桥接与切换使用介绍的详细内容，更多关于Kotlin线程的资料请关注自由互联其它相关文章！