multithreading - 将 channel 桥接到序列

Question

此代码基于 Coroutines guide example: Fan-out

val inputProducer = produce<String>(CommonPool) {
    (0..inputArray.size).forEach {
        send(inputArray[it])
    }
}

val resultChannel = Channel<Result>(10)

repeat(threadCount) {
    launch(CommonPool) {
        inputProducer.consumeEach {
            resultChannel.send(getResultFromData(it))
        }
    }
}

创建 Sequence<Result> 的正确方法是什么？这将提供结果？

Answer 1

您可以获得 channel .iterator() 来自 ReceiveChannel 然后将该 channel 迭代器包装成 Sequence<T> ，实现其正常Iterator<T>阻塞等待每个请求的结果:

fun <T> ReceiveChannel<T>.asSequence(context: CoroutineContext) =
    Sequence {
        val iterator = iterator()
        object : AbstractIterator<T>() {
            override fun computeNext() = runBlocking(context) {
                if (!iterator.hasNext())
                    done() else
                    setNext(iterator.next())
            }
        }
    }

val resultSequence = resultChannel.asSequence(CommonPool)