Deep dive kotlin synergy (18): hot and cold data flow

Series eBook ： Portal

---

Kotlin At first, there was only Channel, But the creator realized that this was not enough . Channel Is a kind of thermal data flow , But we often need cold data flow .

Understanding the difference between hot data flow and cold data flow can help us learn better Flow And other related technologies , Because most of the data sources you use everyday fall into one of these two categories . aggregate （List、Set） It's hot , And sequence and Java Stream It's cold .Channel It's hot , and Flow and RxJava The flow is cold .

Thermal data flow	Cold data flow
Collections(List, Set)	Sequence, Stream
Channel	Flow, RxJava streams

heat vs cold

Hot data flow is urgent 、 Instant , And consumers are independent , Production and storage elements . Cold data flow is lazy , Perform actions on demand , No need to store anything

We can use lists （ heat ） And sequence （ cold ） When I learned about these differences . The builder and operation of hot data flow are started immediately , In the cold data stream , Elements are not produced until they are needed .

@OptIn(ExperimentalStdlibApi::class)
fun main() {
    
    val l = buildList {
    
        repeat(3) {
    
            add("User$it")
            println("L: Added User")
        }
    }

    val l2 = l.map {
    
        println("L: Processing")
        "Processed $it"
    }

    val s = sequence {
    
        repeat(3) {
    
            yield("User$it")
            println("S: Added User")
        }
    }

    val s2 = s.map {
    
        println("S: Processing")
        "Processed $it"
    }
}
// L: Added User
// L: Added User
// L: Added User
// L: Processing
// L: Processing
// L: Processing

therefore , Cold data flow （ Such as Sequence、Stream or Flow）:

• It could be infinite
• Use as few operations as possible
• Use less memory （ There is no need to allocate all intermediate set products ）

Sequence processing does less , Because it delays processing elements . It also works in a very simple way ： Each intermediate operation （ Such as map or filter） Just for decorating the front sequence . The terminal operation is the end point to complete all the work . Please see the following example , Usage sequence ,find Will inquire about map The first qualified element after . It is from sequenceOf（ return 1） Returned sequence , Then map it （1*1 To 1）, And return the result to the filter find. The filter checks whether the element meets its criteria . If the element does not satisfy the condition , The filter will ask again and again . Until you find the right element .

This is very different from the handling of lists , In each intermediate step of list processing, it is necessary to calculate and return a completely processed intermediate set product . This is the difference in the order in which the collection processes elements 、 Need more memory 、 Reasons why more actions may be required （ As shown in the following example ）.

fun m(i: Int): Int {
    
    print("m$i ")
    return i * i
}

fun f(i: Int): Boolean {
    
    print("f$i ")
    return i >= 10
}

fun main() {
    
    listOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
        .map {
     m(it) }
        .find {
     f(it) }
        .let {
     print(it) }
    // m1 m2 m3 m4 m5 m6 m7 m8 m9 m10 f1 f4 f9 f16 16
    
    sequenceOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
        .map {
     m(it) }
        .find {
     f(it) }
        .let {
     print(it) }
    
    // m1 f1 m2 f4 m3 f9 m4 f16 16
}

This means that a list is a collection of elements , The sequence is just the definition of the operator set of the element . Thermal data flow ：

• Ready to use （ Every operation is a terminal operation ）
• You do not need to recalculate the result when using it multiple times

fun m(i: Int): Int {
    
    print("m$i ")
    return i * i
}

fun main() {
    
    val l = listOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
        .map {
     m(it) } // m1 m2 m3 m4 m5 m6 m7 m8 m9 m10
    
    println(l) // [1, 4, 9, 16, 25, 36, 49, 64, 81, 100]
    println(l.find {
     it > 10 }) // 16
    println(l.find {
     it > 10 }) // 16
    println(l.find {
     it > 10 }) // 16
    
    val s = sequenceOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
        .map {
     m(it) }
    
    println(s.toList())
    // [1, 4, 9, 16, 25, 36, 49, 64, 81, 100]
    
    println(s.find {
     it > 10 }) // m1 m2 m3 m4 16
    println(s.find {
     it > 10 }) // m1 m2 m3 m4 16
    println(s.find {
     it > 10 }) // m1 m2 m3 m4 16
}

Java Flow and Kotlin Sequences have the same characteristics , They are all cold data streams .

hot channel, It's cold flow

It's time to get back to the process . establish flow The easiest way is to use a builder , It is similar to produce function , This function is called flow：

val channel = produce {
    
    while (true) {
    
        val x = computeNextValue()
        send(x)
    }
}

val flow = flow {
    
    while (true) {
    
        val x = computeNextValue()
        emit(x)
    }
}

These builders are conceptually equivalent , But because of channel and flow There is a big difference in their behavior , There are also important differences between these two functions . Take a look at the following example ,channel It's hot , So they start counting immediately . This calculation needs to start with a single coroutine . That's why produce Need to be a collaboration builder , It is defined as CoroutineScope Extension function on . The calculation starts immediately , But because the default buffer size is 0（RENDEZVOUS）, So it will be suspended soon , Until the consumer in the following example is ready . Be careful , When there is no receiver , There is a difference between stopping production and producing on demand .channel As a thermal data stream , Generate elements independently of their consumers , And then store them . They don't care how many consumers there are , Because each element can only be received once . After the first receiver consumes all the elements , The second receiver will find an empty 、 It's closed channel, That's why it doesn't receive any elements at all .

private fun CoroutineScope.makeChannel() = produce {
    
    println("Channel started")
    for (i in 1..3) {
    
        delay(1000)
        send(i)
    }
}

suspend fun main() = coroutineScope {
    
    val channel = makeChannel()
    delay(1000)
    println("Calling channel...")
    for (value in channel) {
     println(value) }
    println("Consuming again...")
    for (value in channel) {
     println(value) }
}
// Channel started
// (1 sec)
// Calling channel...
// 1
// (1 sec)
// 2
// (1 sec)
// 3
// Consuming again...

Use Flow It is very different to do the same . Because it is a cold data stream . So production is on demand . It means flow Not a collaboration builder , There is no need for any treatment . It's just a definition of using terminal operations （ Such as collect） The instruction set of how the element should be produced . It runs on the scope that performs its terminal operations （ It starts with the... Of the suspended function continuation Get scope in , It's like coroutineScope Like other coprocessor scoped functions ）, This is flow The builder does not need a CoroutineScope Why .Flow Each terminal operation on the will start from scratch . Please carefully compare the above and the following examples , Because they show channel and flow The key difference between ：

private fun makeFlow() = flow {
    
    println("Flow started")
    for (i in 1..3) {
    
        delay(1000)
        emit(i)
    }
}

suspend fun main() = coroutineScope {
    
    val flow = makeFlow()
    delay(1000)
    println("Calling flow...")
    flow.collect {
     value -> println(value) }
    println("Consuming again...")
    flow.collect {
     value -> println(value) }
}
// (1 sec)
// Calling flow...
// Flow started
// (1 sec)
// 1
// (1 sec)
// 2
// (1 sec)
// 3
// Consuming again...
// Flow started
// (1 sec)
// 1
// (1 sec)
// 2
// (1 sec)
// 3

RxJava Flow and Kotlin Of Flow There are many similarities , Some people even say , Flow It can be called “RxCoroutines”.

summary

Most data sources are either hot or cold ：

• Heat data is urgent , They produce elements as quickly as possible and store them . The elements they create are independent of their consumers , They are collections （List 、Set） and channel
• Cold data flow is inert , They process elements on demand on terminal operations , All intermediate function knowledge defines what should be done （ Usually in decorative mode ）, They usually do not store elements , Instead, create elements as needed , They have very few operations , It could be infinite , They create 、 The process of processing elements is usually next to the consumption process . These elements are Sequence、Java Stream,Flow and RxJava flow （Observable、Single etc. ）

This explains. Channel and Flow The essential difference . It is time to discuss the features supported by the latter .