Kotlin Coroutines - Asynchronously consume a sequence-CodePudding

I'm looking for a way to keep a Kotlin sequence that can produces values very quickly, from outpacing slower async consumers of its values. In the following code, if the async handleValue(it) cannot keep up with the rate that the sequence is producing values, the rate imbalance leads to buffering of produced values, and eventual out-of-memory errors.

getSequence().map { async {
  handleValue(it)
}}

I believe this is a classic producer/consumer "back-pressure" situation, and I'm trying to understand how to use Kotlin coroutines to deal with it.

Thanks for any suggestions :)

CodePudding user response：

Kotlin channels and flows offer buffering producer dispatched data until the consumer/collector is ready to consume it.

But Channels have some concerns that have been manipulated in Flows; for instance, they are considered hot streams:

The producer starts for dispatching data whether or not there is an attached consumer; and this introduces resource leaks.
As long as no consumer attached to the producer, the producer will stuck in suspending state

However Flows are cold streams; nothing will be produced until there is something to consume.

To handle your query with Flows:

GlobalScope.launch {
    flow {
        // Producer
        for (item in getSequence()) emit(item)

    }.map { handleValue(it) }
        .buffer(10) // Optionally specify the buffer size
        .collect { // Collector
            
        }
}