Kotlin流缓冲容量

Question

我有一个关于Kotlin流缓冲能力的问题。以下代码：

import kotlinx.coroutines.flow.*

suspend fun main() = coroutineScope {
    flow { 
        for (i in 1..3) { 
            println("Emiting $i")
            emit(i) 
        }
    }.buffer(0)
    .collect { 
        value -> 
            delay(100)
            println("Consuming $value")
    }
}

生成以下输出：

Emiting 1
Emiting 2
Consuming 1
Emiting 3
Consuming 2
Consuming 3

如果删除缓冲区，结果是：

Emiting 1
Consuming 1
Emiting 2
Consuming 2
Emiting 3
Consuming 3

我是否应该假设，当容量为0时，实际上等于1？

Answer 1

不，容量实际上是0。它看起来就像在缓冲一个元素，因为集合首先消耗一个元素，然后延迟100 is。同时，允许流释放另一个元素。

buffer函数实际上创建了第二个协同线，它允许flow和collect函数并发执行。如果没有对buffer的调用，每个元素必须使用collect才能继续使用下一个元素，因为这两个函数都由相同的协同执行。

让我们逐步了解一下您的代码是如何发生的：

1. 流程延迟1ms，打印Emitting 1，发射1，并挂起。
2. 收集立即消耗1，然后启动100 for的延迟。
3. 流继续，因为1被消耗，延迟1ms，打印Emitting 2，发射2，然后挂起。
4. 收集完成100 of的延迟，打印Consuming 1，然后消耗2，再延迟100 of。
5. 流继续，因为2被消耗，延迟1ms，打印Emitting 3，发射3，然后挂起。
6. 收集完成100 of的延迟，打印Consuming 2，然后消耗3，再延迟100 of。
7. 收集完成延迟100 of，打印Consuming 3，然后完成收集。
8. 流继续，并结束，因为没有更多的元素要发出。

您可以在这里阅读有关buffer的更多信息：https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/buffer.html

Answer 2

不太好，是0。buffer(0)方法使用遮罩下的通道，大小为0有效地使通道成为unbuffered channel。流和使用者都需要准备好发出/消费：

当发送方和接收方相遇时，未缓冲信道传输元素(也称为会合)。缓冲通道

• 流打印“发射1"
• 流试图发出1并挂起(没有缓冲区，读取器也没有等待)
• 收集器准备好阅读了。流发射1，收集器读取1。
• 收集器挂在delay 100上
• 流打印“发射2"
• 流试图发出2并挂起(没有缓冲区，读取器也没有等待)
• 收集器打印“消费1"
• 收集器准备好阅读了。流发射2，收集器读取2。
• 收集器挂在delay 100上
• 流打印“发射3"
• 流试图发出3并挂起(没有缓冲区，读取器也没有等待)
• 收集器打印“消费2"
• 收集器准备好阅读了。流发射3，收集器读取3
• 收集器挂在delay 100上
• 流出口
• 收集器打印“消费3"并退出

当您删除buffer(0)时，情况就有点不同了。不涉及任何渠道。该流程是一种“冷流”，即它在某种程度上等待消费者拉下一个项目：

• 收集器“订阅”该流。
• Flow打印"Emiting 1"，发射"1“并挂起
• 收集器读取"1"，挂起100 for，打印“消费1"
• 收集器要求下一项
• Flow打印"Emiting 2"，发射"2“并挂起
• ..。