[Coroutine] 코루틴 학습 - 14 (Hot and Cold data sources)

Hot data source

• 컬렉션(List, Set 등..)과 같이 컨슈밍과 상관없이 데이터가 미리 생성되어 프로듀싱 되는 데이터를 hot이라고 한다.
• Channel 또한 hot에 속한다.

Cold data source

• Sequence, Java Stream과 같이 실제로 데이터를 필요로 할 때 프로듀싱하는 것을 cold라고 한다.
• Flow, RxJava streams(Observable, Single 등..)을 cold에 속한다.
• 무한하게 실행될 수 있다.
• 최소한의 연산만 수행할 수 있다.
• 메모리를 즉시 할당할 필요가 없기 때문에 필요한 만큼 최소한의 메모리만 사용이 가능하다.

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

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

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

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

 Sequence는 element 연산을 뒤로 미루기 때문에 필요한 만큼의 최소 연산만 수행할 수 있다. Sequence는 각 중간 연산(map 또는 filter와 같은)이 이전의 시퀀스를 데코레이팅하여 동작한다. 그리고 최종 연산에서 이 모든 연산들이 수행된다. 이러한 특징은 모든 element를 메모리 위에 올려놓고 수행해야하는 컬렉션과 큰 차이점이다.

 

 컬렉션은 중간 연산마다 모든 element를 처리하여 모든 컬렉션 데이터들을 처리하는 방식으로 동작한다. 이는 Sequence보다 많은 메모리를 필요로 하는 이유이다.

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

private 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 ") }

    println()

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

 위 예시를 보면 List 컬렉션은 map 연산이 모든 element에 대해서 수행되지만, Sequence는 lazy하게 동작하여 map 연산이 최소한으로만 동작하는 것을 확인할 수 있다.

 

그렇다면 무조건 Cold data stream이 좋은 것인가?

 그렇지는 않다. hot 데이터는 미리 연산이 되어 있기 때문에 재연산할 필요가 없다는 장점이 있다. 상황에 따라서 필요에 맞게 적절하게 사용하면 된다.

 

Hot channels, Cold flow

 channel를 생성할 때 produce 빌더를 사용할 때와 마찬가지로 flow 또한 빌더를 통해 생성할 수 있다. 해당 빌더는 flow()이다.

 

 Channel과 Flow는 유사해보이지만, Channel은 Hot data stream, Flow는 Cold data stream라는 차이가 있다.

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...")
    channel.consumeEach { value -> println(value) }
    println("Consuming again...")
    channel.consumeEach { value -> println(value) }
}
// Channel started
// Calling channel...
// 1
// 2
// 3
// Consuming again...

• Channel은 hot data stream으로, 호출되면 즉시 값을 연산한다. 이때 연산은 각 코루틴에서 시작된다. 이는 produce가 CoroutineScope의 확장 함수로 코루틴 빌더를 정의해야 해야하는 이유다.
• 위 예시는 디폴트 버퍼 사이즈가 0(rendezvous)이므로 리시버가 컨슘하기 전까지 일시중단된다.
• Channel은 각 element를 컨슘과는 독립적으로 생성하고 유지한다.
• 얼마나 많은 리시버가 있는지는 고려하지 않는다.
• 각 element는 한 번만 수신될 수 있으므로 첫 번째 리시버가 모든 element를 컨슘해버리면 두 번째로 컨슘하는 코루틴은 channel이 이미 비어있고 close되어 있는 상황이 된다.

 

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) }
}
// Calling flow...
// Flow started
// 1
// 2
// 3
// Consuming again...
// Flow started
// 1
// 2
// 3

• Flow는 Cold data stream으로, 실제로 연산이 필요로 할 때 element가 생성된다.
• flow는 element가 어떻게 생성되어야 하는지만 정의한다. 정의된 플로우는 종단 연산이 호출 되었을 때 사용된다.
• 이는 flow 빌더가 CoroutineScope를 필요로하지 않는 이유이다.
• flow는 종단 함수가 실행됐을 때에만 코루틴 스코프 내에서 수행된다.
• Channel과 다르게 각 종단 연산은 element 재소비가 가능하다.

---

참고자료

https://www.amazon.com/Kotlin-Coroutines-Deep-Marcin-Moskala/dp/8396395837