Kotlin in Action - 16장 플로우

📖 16.1 플로우는 연속적인 값의 스트림을 모델링한다

suspend fun createValues(): List<Int> {
    return buildList {
        add(1)
        delay(1.seconds)
        add(2)
        delay(1.seconds)
        add(3)
        delay(1.seconds)
    }
}

fun main() = runBlocking {
    val list = createValues()
    list.forEach {
        log(it)
    }
}

모든 값이 계산 된 후 함수가 값을 반환함.
함수가 실행을 마칠 때까지 기다리지 않고 값을 사용할 수 있도록 비동기적으로 반환하고 싶을 때 플롱우가 유용하다.
플로우는 시간이 지남에 따라 나타나는 값과 작업할 수 있게 해주는 코루틴 기반의 추상화다

🔖 16.1.1 플로우를 사용하면 배출되자마자 원소를 처리할 수 있다

fun createValues(): Flow<Int> {
    return flow {
        emit(1)
        delay(1.seconds)
        emit(2)
        delay(1.seconds)
        emit(3)
        delay(1.seconds)
    }
}

fun main() = runBlocking {
    val myFlowOfValues = createValues()
    myFlowOfValues.collect { log(it) }
}

0 [main @coroutine#1] 1
1036 [main @coroutine#1] 2
2042 [main @coroutine#1] 3

원소가 배출되는 즉시 표시된다.

🔖 16.1.2 코틀린 플로우의 여러 유형

콜드 플로우
- 비동기 데이터 스트림
- 값이 실제로 소비되기 시작할 때만 값을 배출
핫 플로우
- 브로드캐스트 방식
- 값이 실제로 소비되고 있는지와 상관없이 독립적으로 배출

📖 16.2 콜드 플로우

🔖 16.2.1 flow 빌더 함수를 사용해 콜드 플로우 생성

fun main() = runBlocking {
    val letters = flow {
        log("Emitting A!")
        emit("A")
        delay(200.milliseconds)
        log("Emitting B!")
        emit("B")
    }
}

아무런 출력도 나타나지 않는다.
flow 빌더 함수를 호출해도 실제 작업이 시작되지 않음.

val counterFlow = flow {
    var x = 0
    while (true) {
        emit(x++)
        delay(200.milliseconds)
    }
}

이 루프는 실제로 플로우가 수집될 때만 실행

🔖 16.2.2 콜드 플로우는 수집되기 전까지 작업을 수행하지 않는다

Flow에 대해 collect 함수를 호출하면 그 로직이 실행된다
플로우를 수집할 때는 플로우 내부의 일시 중단 코드를 실행하므로 collect는 일시 중단 함수이며, 플로우가 끝날 때까지 일시 중단된다.

val letters = flow {
    log("Emitting A!")
    emit("A")
    delay(200.milliseconds)
    log("Emitting B!")
    emit("B")
}

fun main() = runBlocking {
    letters.collect {
        log("Collecting: $it")
        delay(500.milliseconds)
    }
}

0 [main @coroutine#1] Emitting A!
15 [main @coroutine#1] Collecting: A
737 [main @coroutine#1] Emitting B!
737 [main @coroutine#1] Collecting: B

수집자가 플로우의 로직을 실행하는 책임이 있다.
원소 A, B 사이의 지연시간은 700밀리초이다.
- 수집자가 플로우 빌더에 정의된 로직의 실행을 촉발해서 첫 번째 배출 발생
- 수집자가 연결된 람다가 호출되면서 메시지를 기록하고 500밀리초 동안 지연
- 플로우 람다가 계속 실행되며 200밀리초 동안 추가 지연과 배출이 발생

fun main() = runBlocking {
    letters.collect {
        log("(1) Collecting: $it")
        delay(500.milliseconds)
    }
    letters.collect {
        log("(2) Collecting: $it")
        delay(500.milliseconds)
    }
}

0 [main @coroutine#1] Emitting A!
16 [main @coroutine#1] (1) Collecting: A
743 [main @coroutine#1] Emitting B!
743 [main @coroutine#1] (1) Collecting: B
1247 [main @coroutine#1] Emitting A!
1247 [main @coroutine#1] (2) Collecting: A
1958 [main @coroutine#1] Emitting B!
1958 [main @coroutine#1] (2) Collecting: B

같은 플로우를 여러 번 수집할 수 있다.

🔖 16.2.3 플로우 수집 취소

fun main() = runBlocking {
    val collector = launch {
        counterFlow.collect {
            println(it)
        }
    }
    delay(5.seconds)
    collector.cancel()
}

수집자의 코루틴을 취소하면 다음 취소 지점에서 플로우의 수집이 중단된다.

🔖 16.2.4 콜드 플로우의 내부 구현

코틀린의 콜드 플로우는 일시 중단 함수와 수신 객체 지정 람다를 결합한 똑똑한 조합이다.
콜드 플로우의 정의는 매우 간단하다.
- Flow와 FlowCollector라는 2가지 인터페이스만 필요
collect를 호출하면 플로우 빌더 함수의 본문이 실행
이 코드가 emit을 호출하면 emit에 전달된 파라미터로 collect에 전달된 람다가 호출
람다 표현식이 실행을 완료하면 함수는 빌더 함수의 본문으로 돌아가 계속 실행
콜드 플로우는 값 스트림을 처리하는 가벼우면서도 아주 쓸모 있고 확장성 있는 추상화를 제공

🔖 16.2.5 채널 플로우를 사용한 동시성 플로우

suspend fun getRandomNumber(): Int {
    delay(500.milliseconds)
    return Random.nextInt()
}

val randomNumbers = flow {
    repeat(10) {
        emit(getRandomNumber())
    }
}

fun main() = runBlocking {
    randomNumbers.collect {
        log(it)
    }
}

0 [main @coroutine#1] -1441864547
513 [main @coroutine#1] 1101626017
1020 [main @coroutine#1] 313359960
1527 [main @coroutine#1] -2002910241
2033 [main @coroutine#1] 890150628
2538 [main @coroutine#1] -1433970862
3045 [main @coroutine#1] 1503636281
3548 [main @coroutine#1] -1528949568
4055 [main @coroutine#1] -1562012272
4561 [main @coroutine#1] 2076525050

플로우는 순차적으로 실행되며, 모든 계산은 동일 코루틴에서 실행

val randomNumbers = channelFlow {
    repeat(10) {
        launch {
            send(getRandomNumber())
        }
    }
}

동시성으로 호출할 수 있는 플로우 빌더: channelFlow
채널 플로우는 내부적으로 또 다른 동시성 기본 요소인 채널을 관리해야 하기 때문에 생성하는 데 약간 비용이 든다.
- 채널은 코루틴 간 통신을 위한 비교적 저수준의 추상화
플로우 안에서 새로운 코루틴을 시작해야 하는 경우에만 채널 플로우 선택

📖 16.3 핫 플로우

핫 플로우에서는 각 수집자가 플로우 로직 실행을 독립적으로 촉발하는 대신, 여러 구독자라고 불리는 수집자들이 배출된 항목을 공유한다.
- 공유 플로우: 값을 브로드캐스트하기 위해 사용
- 상태 플로우: 상태를 전달하는 특별한 경우에 사용

🔖 16.3.1 공유 플로우는 값을 구독자에게 브로드캐스트한다

공유 플로우는 구독자가 존재하는지 여부에 상관없이 배출이 발생하는 브로드캐스트 방식

class RadioStation {
    private val _messageFlow = MutableSharedFlow<Int>()
    val messageFlow = _messageFlow.asSharedFlow()

    fun beginBroadcasting(scope: CoroutineScope) {
        scope.launch {
            while (true) {
                delay(500.milliseconds)
                val number = Random.nextInt(0..10)
                log("Emitting $number!")
                _messageFlow.emit(number)
            }
        }
    }
}

플로우 빌더를 사용하는 대신 가변적인 플로우에 대한 참조를 얻는다.
배출이 구독자 유무와 관계없이 발생하므로 여러분이 실제 배출을 수행하는 코루틴을 시작할 책임이 있다.

fun main() = runBlocking {
    RadioStation().beginBroadcasting(this)
}

구독자가 없어도 브로드캐스트가 즉시 시작

fun main() = runBlocking {
    val radioStation = RadioStation()
    radioStation.beginBroadcasting(this)
    delay(600.milliseconds)
    radioStation.messageFlow.collect {
        log("A collecting: $it!")
    }
}

약간의 지연 후에 공유 플로우 구독

🛠️ 구독자를 위한 값 재생

공유 플로우 구독자는 구독을 시작한 이후에 배출된 값만 수신

private val _messageFlow = MutableSharedFlow<Int>(replay = 5)

이전에 배출된 원소도 수신하기를 원한다면 replay 파라미터 설정

🛠️ shareIn으로 콜드 플로우를 공유 플로우로 전환

fun querySensor(): Int = Random.nextInt(-10..30)

fun getTemperatures(): Flow<Int> {
    return flow {
        while (true) {
            emit(querySensor())
            delay(500.milliseconds)
        }
    }
}

일정한 간격으로 값을 반환

fun celsiusToFahrenheit(celsius: Int) = celsius * 9.0 / 5.0 + 32.0

fun main() {
    val temps = getTemperatures()
    runBlocking {
        launch {
            temps.collect {
                log("$it Celsius")
            }
        }
        launch {
            temps.collect {
                log("${celsiusToFahrenheit(it)} Fahrenheit")
            }
        }
    }
}

여러번 호출하려면 각 수집자가 센서에 독립적으로 질의

fun main() {
    val temps = getTemperatures()
    runBlocking {
        val sharedTemps = temps.shareIn(this, SharingStarted.Lazily)
        launch {
            sharedTemps.collect {
                log("$it Celsius")
            }
        }
        launch {
            sharedTemps.collect {
                log("${celsiusToFahrenheit(it)} Fahrenheit")
            }
        }
    }
}

shareIn 함수를 사용하면 주어진 콜드 플로우를 한 플로우인 공유 플로우로 변환할 수 있다.
두번째 파라미터 started는 플로우가 실제로 언제 시작돼야 하는지를 정의
- Eagerly는 플로우 수집을 즉시 시작
- Lazily는 첫 번째 구독자가 나타나야만 수집 시작
- WhileSubscribed는 첫 번째 구독자가 나타나야 수집을 시작하고, 마지막 구독자가 사라지면 플로우 수집을 취소
코틀린에서는 시간이 지남에 따라 여러 값을 계산하는 작업을 단순한 콜드 플로우로 노출하고, 필요할 때 콜드 플로우를 핫 플로우로 변환하는 패턴이 자주 사용된다.

🔖 16.3.2 시스템 상태 추적: 상태 플로우

class ViewCounter {
    private val _counter = MutableStateFlow(0)
    val counter = _counter.asStateFlow()
    fun increment() {
        _counter.update { it + 1 }
    }
}

fun main() {
    val vc = ViewCounter()
    vc.increment()
    println(vc.counter.value)
}

값을 배출하는 emit을 사용하는 대신, 값을 갱신하는 update 함수 사용

🛠️ UPDATE 함수로 안전하게 상태 플로우에 쓰기

    fun increment() {
    _counter.value++
}

value 는 가변 속성이니까 위와 같이 구현해도 될까?
위 연산은 원자적이지 않다.
즉, 코루틴들이 여러 스레드에서 실행되기 때문에 문제가 있다.
- 어느 한쪽의 증가 연산이 무효화

🛠️ 상태 플로우는 값이 실제로 달라졌을 때만 값을 배출한다: 동등성 기반 통합

enum class Direction { LEFT, RIGHT }

class DirectionSelector {
    private val _direction = MutableStateFlow(Direction.LEFT)
    val direction = _direction.asStateFlow()

    fun turn(d: Direction) {
        _direction.update { d }
    }
}

fun main() = runBlocking {
    val switch = DirectionSelector()
    launch {
        switch.direction.collect {
            log("Direction now $it")
        }
    }
    delay(200.milliseconds)
    switch.turn(Direction.RIGHT)
    delay(200.milliseconds)
    switch.turn(Direction.LEFT)
    delay(200.milliseconds)
    switch.turn(Direction.LEFT)
}

0 [main @coroutine#2] Direction now LEFT
210 [main @coroutine#2] Direction now RIGHT
405 [main @coroutine#2] Direction now LEFT

LEFT 인자가 한번만 호출이 된다.
상태 플로우가 동등성 기반 통합을 수행하기 때문
- 값이 실제로 달라졌을 때만 구독자에게 값을 배출

🛠️ stateIn 으로 콜드 플로우를 상태 플로우로 변환하기

fun main() = runBlocking {
    val temps = getTemperatures()

    runBlocking {
        val tempState = temps.stateIn(this)
        println(tempState.value)
        delay(800.milliseconds)
        println(tempState.value)
    }
}

stateIn으로 콜드 플로우를 상태 플로우로 변환할 수 있다.
원래 플로우에서 배출된 최신 값을 항상 읽을 수 있다.

🔖 16.3.3 상태 플로우와 공유 플로우의 비교

일반적으로 상태 플로우는 공유 플로우보다 더 간단한 API를 제공

class Broadcaster {
    private val _messages = MutableSharedFlow<String>()
    val messages = _messages.asSharedFlow()
    fun beginBroadcasting(scope: CoroutineScope) {
        scope.launch {
            _messages.emit("Hello!")
            _messages.emit("Hi!")
            _messages.emit("Hola!")
        }
    }
}

fun main(): Unit = runBlocking {
    val broadcaster = Broadcaster()
    broadcaster.beginBroadcasting(this)
    delay(200.milliseconds)
    broadcaster.messages.collect {
        println("Message: $it")
    }
}

첫 번째 구독자가 나타나기 전에 모든 메시지를 배출하는 브로드캐스트
메시지를 출력하지 않음.

class Broadcaster {
    private val _messages = MutableStateFlow<List<String>>(emptyList())
    val messages = _messages.asStateFlow()
    fun beginBroadcasting(scope: CoroutineScope) {
        scope.launch {
            _messages.update { it + "Hello!" }
            _messages.update { it + "Hi!" }
            _messages.update { it + "Hola!" }
        }
    }
}

fun main() = runBlocking {
    val broadcaster = Broadcaster()
    broadcaster.beginBroadcasting(this)
    delay(200.milliseconds)
    println(broadcaster.messages.value)
}

상태 플로우는 전체 메시지 기록을 리스트로 저장하면서 구독자가 모든 이전 메시지에 쉽게 접근할 수 있게 할 수 있다.

🔖 16.3.4 핫 플로우, 콜드 플로우, 공유 플로우, 상태 플로우: 언제 어떤 플로우를 사용할까?

표 16.1 콜드 플로우와 핫 플로우 비교

콜드 플로우	핫 플로우
기본적으로 비활성(수집자에 의해 활성화됨)	기본적으로 활성화됨
수집자가 하나 있음	여러 구독자가 있음
수집자는 모든 배출을 받음	구독자는 구독 시작 시점부터 배출을 받음
보통은 완료됨	완료되지 않음
하나의 코루틴에서 배출 발생(channelFlow 사용 시 예외)	여러 코루틴에서 배출할 수 있음

Kotlin in Action - 16장 플로우

Tags

📖 16.1 플로우는 연속적인 값의 스트림을 모델링한다

🔖 16.1.1 플로우를 사용하면 배출되자마자 원소를 처리할 수 있다

🔖 16.1.2 코틀린 플로우의 여러 유형

📖 16.2 콜드 플로우

🔖 16.2.1 flow 빌더 함수를 사용해 콜드 플로우 생성

🔖 16.2.2 콜드 플로우는 수집되기 전까지 작업을 수행하지 않는다

🔖 16.2.3 플로우 수집 취소

🔖 16.2.4 콜드 플로우의 내부 구현

🔖 16.2.5 채널 플로우를 사용한 동시성 플로우

📖 16.3 핫 플로우

🔖 16.3.1 공유 플로우는 값을 구독자에게 브로드캐스트한다

🛠️ 구독자를 위한 값 재생

🛠️ shareIn으로 콜드 플로우를 공유 플로우로 전환

🔖 16.3.2 시스템 상태 추적: 상태 플로우

🛠️ UPDATE 함수로 안전하게 상태 플로우에 쓰기

🛠️ 상태 플로우는 값이 실제로 달라졌을 때만 값을 배출한다: 동등성 기반 통합

🛠️ stateIn 으로 콜드 플로우를 상태 플로우로 변환하기

🔖 16.3.3 상태 플로우와 공유 플로우의 비교

🔖 16.3.4 핫 플로우, 콜드 플로우, 공유 플로우, 상태 플로우: 언제 어떤 플로우를 사용할까?

표 16.1 콜드 플로우와 핫 플로우 비교