5
我正在生产项目,消耗多个协同例程并推回resultChannel。制片人在最后一项之后关闭频道。扇出/扇入式结果通道
由于resultChannel永远不会关闭,所以代码永远不会结束。如何检测并正确完成迭代,以便hasNext()
返回false
?
val inputData = (0..99).map { "Input$it" }
val threads = 10
val bundleProducer = produce<String>(CommonPool, threads) {
inputData.forEach { item ->
send(item)
println("Producing: $item")
}
println("Producing finished")
close()
}
val resultChannel = Channel<String>(threads)
repeat(threads) {
launch(CommonPool) {
bundleProducer.consumeEach {
println("CONSUMING $it")
resultChannel.send("Result ($it)")
}
}
}
val iterator = object : Iterator<String> {
val iterator = resultChannel.iterator()
override fun hasNext() = runBlocking { iterator.hasNext() }
override fun next() = runBlocking { iterator.next() }
}.asSequence()
println("Starting interation...")
val result = iterator.toList()
println("finish: ${result.size}")
的我发现的方法就是在结果序列中取(100),但我不确定它离开底层结构的状态。 – atok