kotlin - Kotlin 中流发出的并行处理值

Question

Kotlin 代码

runBlocking {
    flow {
        for (i in 0..4) {
            println("Emit $i")
            emit(i)
        }}  .onEach { if (it%2 == 0) delay(200) // Block 1
                println("A: got $it")
            }
            .onEach { println("B: got $it") } // Block 2
            .collect()
}

在控制台打印:

Emit 0
A: got 0
B: got 0
Emit 1
A: got 1
B: got 1
Emit 2
...

在一半的情况下，如何同时运行 block 1 和 block 2 以在 block 1 之前从 block 2 获取消息？

Answer 1

您可以尝试在这些 block 中启动单独的协程:

private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())

flow {
    for (i in 0..4) {
        println("Emit $i")
        emit(i)
    }
}.onEach { // Block 1
    scope.launch {
        if (it % 2 == 0) delay(200) 
        println("A: got $it")
    }

}.onEach { // Block 2
    scope.launch { 
        println("B: got $it") 
    }
}.launchIn(scope)

在这种情况下，Block 2 将在 Block 1 之前打印。 SupervisorJob 需要在此处防止在其中一个协程失败时取消已启动的协程。

此解决方案不保证顺序，例如可以有下一个顺序的日志:

Emit 0
Emit 1
Emit 2
Emit 3
Emit 4

B: got 0
A: got 1
B: got 1
B: got 2
A: got 3
B: got 3
B: got 4
A: got 0
A: got 2
A: got 4