Kotlin - 协程范围，为什么我的异步没有被执行？

Question

CoroutineScopes 是如何工作的？

假设我有一个

enum class ConceptualPosition{
    INVALID,
    A,B
}

假设我有一个用户可以点击任一位置的 UI，A 或 B。

我现在想要一个接收用户输入但在实际请求输入之前忽略它的 Actor。为简单起见，假设只有一种请求职位的方法。

sealed class PositionRequest{
    /**report the next position offered*/
    object ForwardNext:PositionRequest()
}

所以我们可以这样构造：

fun CoroutineScope.positionActor(
        offeredPosition:ReceiveChannel<ConceptualPosition>,
        requests:ReceiveChannel<PositionRequest>,
        output:SendChannel<ConceptualPosition>
) = launch{
    var lastReceivedPosition = INVALID
    var forwardNextReceived = 0

    println("ACTOR: entering while loop")
    while(true) {
        select<Unit> {
            requests.onReceive {
                println("ACTOR: requests.onReceive($it)")
                when (it) {
                    is PositionRequest.ForwardNext -> ++forwardNextReceived
                }
            }

            offeredPosition.onReceive {
                println("ACTOR: offeredPosition.onReceive($it)")
                lastReceivedPosition = it
                if (forwardNextReceived > 0) {
                    --forwardNextReceived
                    output.send(it)
                }
            }
        }
    }
}

然后构建一个门面与之交互：

class BasicUI{
    private val dispatcher = Dispatchers.IO

    /*start a Position Actor that receives input from the UI and forwards them on demand*/
    private val requests = Channel<PositionRequest>()
    private val offeredPositions = Channel<ConceptualPosition>()
    private val nextPosition = Channel<ConceptualPosition>()
    init {
        runBlocking(dispatcher){
            positionActor(offeredPositions,requests,nextPosition)
        }
    }

    /** Receives a [ConceptualPosition] that may or may not get accepted and acted upon.*/
    fun offerPosition(conceptualPosition: ConceptualPosition) = runBlocking(dispatcher) {
        offeredPositions.send(conceptualPosition)
    }

    /** waits for a [ConceptualPosition] to be offered via [offerPosition], then accepts it*/
    fun getPosition(): ConceptualPosition = runBlocking(dispatcher){
        requests.send(PositionRequest.ForwardNext)
        nextPosition.receive()
    }
}

这当然行不通，因为runBlocking 是CoroutineScope，init 将不会返回，直到positionActor(offeredPositions,requests,nextPosition) 启动的协程结束......这永远不会因为while(true) 在它。

那么如果我们让BasicUI 实现CoroutineScope 呢？毕竟，那个是what Roman Elizarov said we should do at the KotlinConf，如果我理解正确的话，应该将positionActor(...)创建的协程绑定到BasicUI实例，而不是runBlocking-block。

让我们看看...

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlin.coroutines.CoroutineContext

class BasicUI:CoroutineScope{

    private val dispatcher = Dispatchers.IO

    private val job = Job()
    override val coroutineContext: CoroutineContext
        get() = job

    /*start a Position Actor that receives input from the UI and forwards them on demand*/
    private val requests = Channel<PositionRequest>()
    private val offeredPositions = Channel<ConceptualPosition>()
    private val nextPosition = Channel<ConceptualPosition>()
    init {
        positionActor(offeredPositions,requests,nextPosition)
    }

    /** Receives a [ConceptualPosition] that may or may not get accepted and acted upon.*/
    fun offerPosition(conceptualPosition: ConceptualPosition) = runBlocking(dispatcher) {
        offeredPositions.send(conceptualPosition)
    }

    /** waits for a [ConceptualPosition] to be offered via [offerPosition], then accepts it*/
    fun getPosition(): ConceptualPosition = runBlocking(dispatcher){
        requests.send(PositionRequest.ForwardNext)
        nextPosition.receive()
    }
}

让我们构建一个小测试用例：我将提供一些他应该忽略的As，然后启动一个持续提供Bs 的协程，其中一个会在我询问时返回给我演员。

import ConceptualPosition.*
import kotlinx.coroutines.async
import kotlinx.coroutines.runBlocking

fun main(args: Array<String>) = runBlocking{
    val ui = BasicUI()
    println("actor engaged")

    //these should all be ignored
    repeat(5){ui.offerPosition(A)}
    println("offered some 'A's")

    //keep offering 'B' so that eventually, one will be offered after we request a position
    async { while(true){ui.offerPosition(B)} }

    //now get a 'B'
    println("requesting a position")
    val pos = ui.getPosition()
    println("received '$pos'")
}

这会导致

actor engaged
ACTOR: entering while loop
ACTOR: offeredPosition.onReceive(A)
ACTOR: offeredPosition.onReceive(A)
ACTOR: offeredPosition.onReceive(A)
ACTOR: offeredPosition.onReceive(A)
offered some 'A's
ACTOR: offeredPosition.onReceive(A)
requesting a position
ACTOR: requests.onReceive(PositionRequest$ForwardNext@558da0e9)

...什么也没有。

显然，B 从未被提供——因此，从未被转发——这会导致主线程阻塞（在这种情况下应该如此）。

我扔了一个

if(conceptualPosition == ConceptualPosition.B) throw RuntimeException("B offered?!")

进入BasicUI.offerPosition，没有例外，所以...

在这一点上，我可能不得不承认我还不了解 Kotlin CoroutineScope。

为什么这个例子不起作用？

Answer 1

这里似乎有两个问题：

1. offerPosition/getPosition 不是挂起函数。在大多数情况下，使用runBlocking 是错误的解决方案，应该在必须与同步代码或主函数交互时使用。
2. 在当前CoroutineScope 中执行不带任何参数的async。对于您的主要功能，这是runBlocking。文档实际上描述了这种行为：

此构建器的默认 CoroutineDispatcher 在事件循环的内部实现中处理此阻塞线程中的继续，直到此协程完成。有关 kotlinx.coroutines 提供的其他实现，请参阅 CoroutineDispatcher。

简单来说，async 块不会在事件循环中轮到执行，而其他延续正在使用它。由于getPosition 正在阻塞，因此您阻塞了事件循环。

将阻塞函数替换为挂起函数和withContext(dispatcher) 以在不同的执行器上调度将允许异步函数运行并最终解决状态。