3.3 채널과 플로우 - 핫 데이터 소스와 콜드 데이터 소스

 

 

우리가 사용하는 대부분의 데이터 소스는 핫 스트림 데이터와 콜드 스트림 데이터 두 가지로 구분할 수 있다.

즉, 이 두 가지의 차이를 이해하는 것이 중요하다.

핫	콜드
컬렉션(List, Set)	Sequence, Stream
Channel	FLow, Rx.Java 스트림

(List, Set과 같은) 컬렉션은 핫이고, Sequence와 자바의 Stream은 콜드다.

Channel은 핫이고, Flow와 (Observable, Single과 같은) Rxlava 스트림은 콜드다.

 

 

핫 vs 콜드

핫 데이터 스트림은 열정적이라 데이터를 소비하는 것과 무관하게 원소를 생성한다.

반면에 콜드 데이터 스트림은 게을러서 요청이 있을 때만 작업을 수행하며, 아무것도 저장하지 않는다.

 

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

리스트(핫)와 시퀀스(콜드)를 사용할 때 그 차이가 드러난다.

리스트의 빌더와 연산은 즉각 실행된다. 시퀀스는 원소가 실제로 필요할 때까지 빌더와 연산이 실행되지 않는다.

 

즉, 콜드 데이터 스트림은 (Sequence, Stream, Flow)

• 무한할 수 있다.
• 최소한의 연산만 수행한다.
• (중간에 생성되는 값들을 저장하지 않기 때문에) 메모리를 적게 사용한다.

중간 연산(map, filter 등)은 이전 시퀀스에 새로운 연산을 추가할 뿐이다.

최종 연산이 호출되기 전까지는 아무런 연산을 수행하지 않고, 최종 연산에서 모든 작업을 실행한다.

 

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) }
    // m m2 m3 m4 m5 m6 m7 m8 mg m10 f1 f4 f9 f16 16

    println()

    sequenceOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
        .map { m(it) }
        .find { f(it) }
        .let { print(it) }
    // m f1 m2 f4 m3 19 m4 f16 16
}

위 예제에서 시퀀스의 find는 map에게 첫 번째 원소를 달라고 한다.

map은 sequenceOf에게 첫 번째 원소를 받고, map을 수행한 뒤 find에 넘긴다.

find는 위에서 받은 원소(1)가 요구 사항에 부합하는지 확인한다.

요구사항을 만족하지 못한다면, find는 적절한 원소를 찾을 때까지 계속해서 질의한다.

 

반면에 리스트는 중간 과정에서 모든 원소에 대한 계산을 수행하고, 결과가 담긴 컬렉션을 각각 생성한다.

따라서 리스트는 원소의 처리 순서도 다르고, 컬렉션 처리 과정에서 좀더 많은 메모리를 필요로 하며, 더 많은 연산을 수행한다.

 

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) } // ml 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 }) // ml m2 m3 m4 16
}

리스트는 원소를 가지고 있는 컬렉션이지만, 시퀀스는 원소를 어떻게 계산할지 정의한 것 뿐이다.

핫 데이터 스트림은

• 항상 사용 가능한 상태다. (각 연산이 최종 연산이 될 수 있다)
• 여러 번 사용되었을 때 매번 결과를 다시 계산할 필요가 없다.

 

 

핫 채널, 콜드 플로우

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

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

다시 코루틴으로 돌아와서, 콜드 데이터 스트림인 플로우를 생성해 보자.

플로우를 생성하는 가장 일반적인 방법은 produce 함수와 비슷한 flow를 사용하는 것이다.

 

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초 후)
Calling channel...
1
(1초 후)
2
(1초 후)
3
(1초 후)
Consuming again...

채널은 핫 데이터 스트림이기 때문에 소비되는 것과 상관없이 값을 곧바로 생성한 후, 값을 가지고 있는다.

수신자가 얼마나 많은지는 신경쓰지 않는다.

별도의 코루틴에서 계산을 곧바로 수행한다. 따라서 produce는 CoroutineScope 확장 함수의 코루틴 빌더가 된다.

 

채널은 곧바로 시작되지만, 버퍼의 크기가 0인 랑데뷰인 채널이기 때문에 수신자가 원소를 받을 때까지 중단된다.

각 원소는 단 한 번만 받을 수 있기 때문에, 첫 번째 수신자가 모든 원소를 받으면 두 번째 소비자는 채널이 닫힌 후기 때문에 아무 원소도 받을 수 없다.

 

 

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초 후)
Calling flow...
Flow started
(1초 후)
1
(1초 후)
2
(1초 후)
3
Consuming again...
Flow started
(1초 후)
1
(1초 후)
2
(1초 후)
3

플로우는 콜드 데이터 소스이기 때문에 값이 필요할 때만 생성한다.

flow는 단지 최종 연산이 호출될 때 원소가 어떻게 생성되어야 하는지 정의한 것에 불과하다.

그래서 flow 빌더는 코루틴 빌더가 아니고, 어떠한 처리도 하지 않는다. CoroutineScope가 필요하지 않다.

 

flow 빌더는 최종 연산의 스코프에서 실행된다.

coroutineScope 같은 코루틴 스코프 함수처럼, 중단 함수의 컨티뉴에이션 객체로부터 스코프를 가지고 온다.

 

 

요약

• 핫 데이터 소스는 열정적이다.
  • 가능한 빨리 원소를 만들고 저장하며, 원소가 소비되는 것과 무관하게 생성한다.
  • 예를 들어 컬렉션(List, Set), Channel이 있다.

• 콜드 데이터 소스는 게으르다.
  
  • 일반적으로 원소를 저장하지 않고, 필요할 때 원소를 생성한다.
  • 최종 연산에서 값을 생성한다. 중간 과정에서는 무엇을 해야 할지만 정의한 것이다.
  • 연산은 최소한으로 수행하며, 무한정일 수 있다.
  • 예를 들어 Sequence, 자바 Stream, Flow, RxJava 스트림(Observable, Single)이 있다.

 

 

 

출처

https://www.yes24.com/product/goods/123034354