kotlin协程-runBlocking

runBlocking是kotlin提供的一个协程启动函数，它的作用是运行指定的中断函数，同时保证在runBlocking的协程体执行完成之前,runBlocking的下一行代码不会执行。

在我们阅读runBlocking源码之前，可以思考一下，如果让我们来实现runBlocking,我们应该怎么做，

@InternalCoroutinesApi
class MyStdBlockCoroutine<T>(
    parentContext: CoroutineContext,
    val thread: Thread
) : AbstractCoroutine<T>(parentContext, true, true) {
    override fun afterCompletion(state: Any?) {
        super.afterCompletion(state)
        if (Thread.currentThread() != thread) {
            LockSupport.unpark(thread)
        }
    }

    fun joinSync() {
        while (true) {
            if (Thread.interrupted()) throw InterruptedException().also { cancelCoroutine(it) }
            if (isCompleted) break
            LockSupport.parkNanos(Long.MAX_VALUE)
        }

        val state = state this.state.unboxState()
        (state as? CompletedExceptionally)?.let { throw it.cause }
        return state as T
    }
}

@InternalCoroutinesApi
fun <T> myRunBlocking(block: suspend CoroutineScope.() -> T) {
    val aimContext = GlobalScope.newCoroutineContext(EmptyCoroutineContext)
    val coroutine = MyStdBlockCoroutine<T>(aimContext, Thread.currentThread())
    coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
    return coroutine.joinSync()
}

我这里继承了AbstractCoroutine，并且对外提供了joinSync() 这个方法，这个方法的作用是在状态变更到结束状态之前，会通过LockSupport阻塞当前线程，当协程函数体调用完成之后，会触发afterCompletion调用，这个时候再唤醒线程，就可以达成在runBlocking内部运行的协程未执行完成之后，不执行runBlocking之后的代码，实现在该线程阻塞的效果。

我这样的实现有没有啥问题呢，当然有！如果我想要实现在当前线程执行协程，现在的逻辑肯定不行，因为我使用GlobalScope.newCoroutineContext创建了一个新的上下文，其中使用的默认的调度器会让新协程运行在后台线程。如果我期望协程体默认在当前线程下执行，我应该怎么做？

答案是实现一个CoroutineDispatcher

这个CoroutineDispatcher的核心是一个消息队列，我们知道协程在最终运行的时候会被包装成一个任务，将该任务交给对应的调度器去执行。我们只需要在我们的实现中将该任务加入到队列，然后在合适的地方遍历该队列，就好了，参考上文的demo代码，这个位置应该是joinSync,伪代码如下

while(true) {
  val parkTime = pollAndRunFromQueue();
  if (Thread.interrupted()) throw InterruptedException().also { cancelCoroutine(it) }
  if (isCompleted) break
  LockSupport.parkNanos(parkTime)
}

当然最终的调度器为了实现可以在协程体中调用delay,还需要做一些额外的工作，不过那并不是本文的重点，在我们用自己的方式做了一些思考之后，我们来看一下runBlocking的实现。

public actual fun <T> runBlocking(context: CoroutineContext, block: suspend CoroutineScope.() -> T): T {
    contract {
        callsInPlace(block, InvocationKind.EXACTLY_ONCE)
    }
    val currentThread = Thread.currentThread()
    val contextInterceptor = context[ContinuationInterceptor]
    val eventLoop: EventLoop?
    val newContext: CoroutineContext
    if (contextInterceptor == null) {
        // create or use private event loop if no dispatcher is specified
        eventLoop = ThreadLocalEventLoop.eventLoop
        newContext = GlobalScope.newCoroutineContext(context + eventLoop)
    } else {
        // See if context's interceptor is an event loop that we shall use (to support TestContext)
        // or take an existing thread-local event loop if present to avoid blocking it (but don't create one)
        eventLoop = (contextInterceptor as? EventLoop)?.takeIf { it.shouldBeProcessedFromContext() }
            ?: ThreadLocalEventLoop.currentOrNull()
        newContext = GlobalScope.newCoroutineContext(context)
    }
    val coroutine = BlockingCoroutine<T>(newContext, currentThread, eventLoop)
    coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
    return coroutine.joinBlocking()
}

可以看到如果外层调用runBlocking在没有使用调度器的情况下，会创建一个EventLoop,这个EventLoop的作用就是我上面说描述的自己实现的消息队列，创建完消息队列之后，创建了一个BlockingCoroutine，然后调用start启动协程，该api会把协程代码包装成任务抛到EventLoop中，最后调用joinBlocking()阻塞当前线程，我们来看一下joinBlocking做了什么。

fun joinBlocking(): T {
        registerTimeLoopThread()
        try {
            eventLoop?.incrementUseCount()
            try {
                while (true) {
                    @Suppress("DEPRECATION")
                    if (Thread.interrupted()) throw InterruptedException().also { cancelCoroutine(it) }
                    val parkNanos = eventLoop?.processNextEvent() ?: Long.MAX_VALUE
                    // note: process next even may loose unpark flag, so check if completed before parking
                    if (isCompleted) break
                    parkNanos(this, parkNanos)
                }
            } finally { // paranoia
                eventLoop?.decrementUseCount()
            }
        } finally { // paranoia
            unregisterTimeLoopThread()
        }
        // now return result
        val state = this.state.unboxState()
        (state as? CompletedExceptionally)?.let { throw it.cause }
        return state as T
    }

可以看到在joinBlocking中有eventLoop?.processNextEvent()这个调用，它的核心逻辑就是从消息队列中执行下一个任务，然后这个方法会返回需要阻塞线程的时间。比如下一个任务是需要delay的，就会出现这个场景。除此之外我们看到在joinBlocking开头有一个incrementUseCount(),在finally中有一个decrementUseCount(),看上去是引用计数相关的，实际分析代码也证明了这点，incrementUseCount会将引用计数+1,decrementUseCount()会将引用计数-1，当队列的引用计数为0时，触发队列的清理功能。那么为什么需要这个？原因是kotlin官方还考虑到了runBlocking嵌套调用的场景，这种情况下如果有delay的任务存在，最好还是放在同一个队列里去操作，放在不同的队列里操作可能会导致joinBlocking中的返回的停止当前线程的时长不一致，容易产生bug.从这点来看，只能说不愧是标准库。

joinBlocking的最后一步就是将状态解包成数据。然后将这个数据返回。
经过对runBlocking的分析，结合我自己的思考，这个方法在你需要桥接协程代码到同步代码中的时候是很有用的，但这个方法是不应该在主线程中调用的，只适用于后台线程的场景。