Flow를 공부해보자 3편

Yerin·2023년 11월 16일

Flow

목록 보기

4/6

오늘도 공식문서 읽고 ..
거기에 스터디 내용을 첨부하는 ..

저번 스터디에서도 대화했던 buffer 라는 개념에 대해 확실하게 하고 가야할듯 ~

Flow Context

Flow는 항상 호출한 코루틴의 context에서 이루어진다. 이러한 Flow의 속성을 context preservation (콘텍스트 보존)이라고 부른다.

기본적으로 flow builder 내의 코드들은 해당 flow의 collector가 제공하는 context에서 실행된다.

A common pitfall when using withContext

보통 CPU를 소비하는 시간이 긴 코드는 Dispatcher.Default의 context,

UI를 업데이트하는 코드는 Dispatcher.Main의 context에서 수행한다.

코틀린 코루틴에서는 context를 변경하기 위해 주로 withContext를 사용하는데, flow builder 내에서는 콘텍스트 보존을 준수해야하며, 다른 context에서 값을 방출하는 것이 허용되지 않는다.

fun simple(): Flow<Int> = flow {
    // The WRONG way to change context for CPU-consuming code in flow builder
    kotlinx.coroutines.withContext(Dispatchers.Default) {
        for (i in 1..3) {
            Thread.sleep(100) // pretend we are computing it in CPU-consuming way
            emit(i) // emit next value
        }
    }
}

</br>

fun main() = runBlocking<Unit> {
    simple().collect { value -> println(value) } 
}

예를 들어, 위 코드를 실행하면 다음과 같은 오류가 발생한다.

Exception in thread "main" java.lang.IllegalStateException: Flow invariant is violated:
		Flow was collected in [CoroutineId(1), "coroutine#1":BlockingCoroutine{Active}@5511c7f8, BlockingEventLoop@2eac3323],
		but emission happened in [CoroutineId(1), "coroutine#1":DispatchedCoroutine{Active}@2dae0000, Dispatchers.Default].
		Please refer to 'flow' documentation or use 'flowOn' instead
	at ...

flowOn operator

위의 예제 오류를 보면 flowOn을 대신 사용하라고 말하고 있다.
context를 바꾸고 싶다면 다음과 같이 코드를 작성해야한다.

fun simple(): Flow<Int> = flow {
    for (i in 1..3) {
        Thread.sleep(100) // pretend we are computing it in CPU-consuming way
        log("Emitting $i")
        emit(i) // emit next value
    }
}.flowOn(Dispatchers.Default) // RIGHT way to change context for CPU-consuming code in flow builder

fun main() = runBlocking<Unit> {
    simple().collect { value ->
        log("Collected $value") 
    } 
}

main 스레드에서 collection이 발생하는 동안, flow {..} 코드는 background에서 돌아간다.

원래라면 flow에서 하나의 코루틴이 emit과 collect를 순차적으로 하던 것을
flowOn을 이용하면 각각 다른 코루틴에서 collection와 emit이 발생하게 된다.

flowOn 연산자는 코루틴 디스패처를 변경할 때, upstream에게 새로운 코루틴을 생성해준다.

flowOn

flowOn은 buffer 연산자를 명시적으로 호출하지는 않지만, buffer와 같이 동작한다.

Buffering

다른 코루틴에서 flow를 실행하는 것은 flow를 collect하는 전체 시간의 관점에서는 이득이다. (특히 긴 시간의 비동기 작업이 포함되어있을 때)

fun simple(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100) // pretend we are asynchronously waiting 100 ms
        emit(i) // emit next value
    }
}

fun main() = runBlocking<Unit> { 

    val time = measureTimeMillis {
        simple()
            .buffer() // buffer emissions, don't wait
            .collect { value -> 
                delay(300) // pretend we are processing it for 300 ms
                println(value) 
            } 
    }   
    println("Collected in $time ms")

}

buffer 연산자를 사용하면 collect와 emit를 순차적이 아닌 동시에 실행시킬 수 있다.

위의 예시코드는 처음 100ms만 기다리고, 다음부터는 방출을 기다리지 않아도 된다.

Conflation

flow를 통해 방출되는 모든 값을 필요로하는 것이 아니라, 최근의 하나만 필요할 때가 있다. 이 경우에는 conflation 연산자를 사용해보자. 이 연산자를 사용하면 collector가 너어무 느린경우에 중간 방출 값들은 무시할 수 있다.

fun simple(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100) // pretend we are asynchronously waiting 100 ms
        emit(i) // emit next value
    }
}

fun main() = runBlocking<Unit> { 
    val time = measureTimeMillis {
        simple()
            .conflate() // conflate emissions, don't process each one
            .collect { value -> 
                delay(300) // pretend we are processing it for 300 ms
                println(value) 
            } 
    }   
    println("Collected in $time ms")
}

위의 예제 코드를 실행해보면 다음과 같다.

1
3
Collected in 758 ms

Processing the latest value

conflation은 emitter와 collector가 모두 생산 속도가 느릴 때 사용 가능한 방법이다. 이 방법은 방출된 값을 무시하는 방법으로 이뤄진다.

다른 방법으로는, slow collector를 취소하고, 새로운 값이 방출 될 때마다 collector를 재실행하는 방법이 있다.

fun simple(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100) // pretend we are asynchronously waiting 100 ms
        emit(i) // emit next value
    }
}

fun main() = runBlocking<Unit> { 
    val time = measureTimeMillis {
        simple()
            .collectLatest { value -> // cancel & restart on the latest value
                println("Collecting $value") 
                delay(300) // pretend we are processing it for 300 ms
                println("Done $value") 
            } 
    }   
    println("Collected in $time ms")
}

🤔 여기서 내가 들었던 의문 ..
collectLatest는 마지막 방출 값이라는 걸 어떻게 아는걸까..?

마지막 값인걸 판단하는게 아니라 delay시간이 끝날때까지 값이 안와서 latest로 찍히는 거임

delay에 걸리면 collectLatest { }를 다시 타게 된다.

위 코드를 실행하면 다음과 같다.

Collecting 1
Collecting 2
Collecting 3
Done 3
Collected in 741 ms

스터디 하면서 추가로 몇가지 알게된 지식

Lifecycle 안에서 collect문은 하나밖에 등록 못한다
-> 내부적으로 무한 루프를 돌기 때문.. (이거는 나중에 더 자세히 알아봐야할 듯)
Buffer 더 깊이 보기 : 공식문서

fun <T> Flow<T>.buffer(
    capacity: Int = BUFFERED, 
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND
): Flow<T>

capacity : buffer size 정함 -> BUFFERED, CONFLATED
onBufferOverFlow : 기본값이 BufferOverFlow.SUSPEND

Yerin

𝙸 𝚐𝚘𝚝𝚝𝚊 𝚕𝚒𝚟𝚎 𝚖𝚢 𝚕𝚒𝚏𝚎 𝙽𝙾𝚆, 𝙽𝙾𝚃 𝚕𝚊𝚝𝚎𝚛 !