android - Coroutines Channel values consumeEach waits 而不是一次消费

Question

我在弄清楚如何使用 channel 时遇到问题我希望它们在调用发送后立即将值推送给消费者，而不是在加载来自两个源的数据后获取值。

MainActivity.kt

fun loadData() {
    textView.text = "LOADING"
    launch {
        repository.loadData().consumeEach { loaded ->
            withContext(Dispatchers.Main) {
                logd("Presenting: ${loaded.size}, $loaded")
                textView.text = loaded.joinToString { "$it\n" }
            }
        }
    }

Repository.kt

suspend fun loadData(): ReceiveChannel<List<String>> {
    return coroutineScope {
        produce(capacity = 2) {
            launch {
                val localData = local.loadData()
                send(localData)
            }
            launch {
                val remoteData = remote.loadData()
                send(remoteData)
            }
        }
    }
}

Remote.kt

override val data: MutableList<String> = mutableListOf("R1", "R2", "R3", "R4", "R5")

override suspend fun loadData(): List<String> {
    logd("Loading remote started")
    val wait = Random.nextLong(0, 500)
    delay(wait)
    logd("Remote loading took $wait")
    logd("Loading remote finished: ${data.size}, $data")
    return data
}

Local.kt

override val data: MutableList<String> = mutableListOf("L1", "L2", "L3", "L4", "L5")

override suspend fun loadData(): List<String> {
    logd("Loading local started")
    val wait = Random.nextLong(1000, 2000)
    delay(wait)
    logd("Local loading took $wait")
    logd("Loading local finished: ${data.size}, $data")
    return data
}

---

我在控制台中得到了这个

D/Local: Loading local started
D/Remote: Loading remote started
D/Remote: Remote loading took 265
D/Remote: Loading remote finished: 5, [R1, R2, R3, R4, R5]
D/Local: Local loading took 1650
D/Local: Loading local finished: 5, [L1, L2, L3, L4, L5]
D/DispatchedCoroutine: Presenting: 5, [R1, R2, R3, R4, R5]
D/DispatchedCoroutine: Presenting: 5, [L1, L2, L3, L4, L5]

看起来两个来源的数据都是在达到容量后发出的。我希望它做的是消费者可以在发送数据后立即接收数据。所以控制台输出看起来更像这样。

D/Local: Loading local started
D/Remote: Loading remote started
D/Remote: Remote loading took 265
D/DispatchedCoroutine: Presenting: 5, [R1, R2, R3, R4, R5]
D/Remote: Loading remote finished: 5, [R1, R2, R3, R4, R5]
D/Local: Local loading took 1650
D/Local: Loading local finished: 5, [L1, L2, L3, L4, L5]
D/DispatchedCoroutine: Presenting: 5, [L1, L2, L3, L4, L5]

我如何使用 coroutine.Channel 实现这一点(发送后立即消耗值)？

---

编辑#1:

从 Repository#loadData() 中删除 coroutineScope{...} 后，它开始按预期工作。但现在我遇到了一个问题，我必须将范围作为函数参数传递，这对我来说非常难看。

Repository.kt

suspend fun loadData(scope: CoroutineScope): ReceiveChannel<List<String>> {
    return scope.produce(capacity = 2) {
        launch {
            val localData = local.loadData()
            send(localData)
        }
        launch {
            val remoteData = remote.loadData()
            send(remoteData)
        }
        invokeOnClose {
            logd("Closing channel")
        }
    }
}

Answer 1

我认为您的代码看起来不错，可以完成您期望的工作。我认为您遇到的问题是日志记录在发生时没有到达您的控制台。请记住，日志记录本身有自己的缓冲和 IO 线程要经过。我已经尝试了您的代码并改用了 println，并且我得到了您预期的行为。为了确认，您可以不进行随机等待，而是将每次等待的等待时间增加到 10 秒，并真正让它们一个接一个地发生。只是为了帮助您自己确认这一点，这是我的非 Android 版本的您正在尝试做的事情:

    fun main() = runBlocking {
    val start = System.currentTimeMillis()
    launch(Dispatchers.Unconfined) {
        loadData().consumeEach { loaded ->
            println("Presenting: ${loaded.size}, $loaded")
        }
    }.join()
    println("The whole thing took ${System.currentTimeMillis() - start}")
}

suspend fun CoroutineScope.loadData() = produce {
    launch {
        val localData = localloadData()
        send(localData)
    }
    launch {
        val remoteData = remoteloadData()
        send(remoteData)
    }
}

val remoteData: MutableList<String> = mutableListOf("R1", "R2", "R3", "R4", "R5")

suspend fun remoteloadData(): List<String> {
    println("Loading remote started")
    val wait = 500L
    delay(wait)
    println("Remote loading took $wait")
    println("Loading remote finished: ${remoteData.size}, $remoteData")
    return remoteData
}

val localData: MutableList<String> = mutableListOf("L1", "L2", "L3", "L4", "L5")

suspend fun localloadData(): List<String> {
    println("Loading local started")
    val wait = 1000L
    delay(wait)
    println("Local loading took $wait")
    println("Loading local finished: ${localData.size}, $localData")
    return localData
}

它产生这个:

Loading local started
Loading remote started
Remote loading took 500
Loading remote finished: 5, [R1, R2, R3, R4, R5]
Presenting: 5, [R1, R2, R3, R4, R5]
Local loading took 1000
Loading local finished: 5, [L1, L2, L3, L4, L5]
Presenting: 5, [L1, L2, L3, L4, L5]
The whole thing took 1046

编辑:我删除了你用来更新你的值的 withContext(Dispatchers.Main) - 这真的不需要。您已经在这个阶段完成了异步工作。相反，您需要在顶级 launch 中指定上下文，就像现在一样。

除非您另有说明，否则下面的其余工作应继承该上下文。无需继续将上下文作为参数传递。

如果您确实发现自己处于另一个上下文取代继承上下文的位置，那么可能将其作为参数传递可能是一种方法，但我的偏好是找到一个解决方法并以一种方式表达它确实继承自调用上下文。