[Flow Basics 3] Flow Context 핸들링 하는 방법

 

Flow에서 Context는 두 가지 관점에서 볼 수 있다.

1) 데이터를 Collect 하는 부분의 Contex와, 2) Emit하는 부분의 Context이다.

1. Consumer의 Context

일반적으로 Collect 되는 부분의 Context는 코루틴이 호출된 곳의 Context를 따른다.

아래와 같이 withContext를 활용해 어떤 Dispatcher를 활용해 데이터를 collect 할지 개발자가 지정할 수 있다.

Flow의 이러한 성질을 context preservation(콘텍스트 보존)이라 한다.

fun main() = runBlocking {
  withContext(context) { ✅
    simple().collect { value ->
        println(value) // run in the specified context
    }
  }
}

2. Producer의 Context

Flow를 방출하는 Producer의 경우, 디폴트로 Consumer의 Context에서 일어난다.

아래 코드를 실행하게 되면 Collect와 emit 모두 main 스레드에서 실행됨을 확인할 수 있다.

main함수에서 불렸기 Collect가 불렸기 때문에, collect 되는 부분도 main, 그리고 producer도 이에 따라 main이 되는 것이다.

fun simple(): Flow<Int> = flow {
    printWithThread("Started simple flow")
    for (i in 1..3) { emit(i) }
}  
​
fun main() = runBlocking<Unit> {
    simple().collect { value -> printWithThread("Collected $value") } 
}
​
fun printWithThread(str: Any?) {
    println("[${Thread.currentThread().name}] $str")
}
​
// 🖥️ 출력 결과
[main] Started simple flow
[main] Collected 1
[main] Collected 2
[main] Collected 3

 

그럼 다른 코드를 가만히 놔둔 채 main의 코드만 withContext를 활용해 consumer의 context를 바꾸는 경우는 어떻게 될까? 이 경우에도 producer가 consumer의 context에 따라 DefaultDispatcher에서 모든 코드가 실행되게 된다.

fun main() = runBlocking<Unit> {
    withContext(Dispatchers.Default) {
        simple().collect { value -> printWithThread("Collected $value") }
    }
}
​
// 🖥️ 출력 결과
[DefaultDispatcher-worker-1] Started simple flow
[DefaultDispatcher-worker-1] Collected 1
[DefaultDispatcher-worker-1] Collected 2
[DefaultDispatcher-worker-1] Collected 3

 

3. flowOn 맞고요🙆🏻 withContext 아닙니다🙅🏻

 

그렇다면 데이터를 방출하는 곳과 수집하는 곳의 Context를 다르게 하고 싶은 경우에는 어떻게 할까?

Android를 예로 들자면, Producer의 경우 CPU 자원을 많이 잡아먹어서 새로운 Defualt 디스패처를 지정해주고 싶으며, Consumer의 경우 UI를 업데이트하기 위해 UI스레드를 지정해야 할 경우가 될 수 있겠다!

이럴 때는 flowOn연산자를 활용하여 Producer의 Context를 지정해 줄 수 있다.

fun rightExample(): Flow<Int> = flow {
    for (i in 1..3) {
        Thread.sleep(100) // pretend we are computing it in CPU-consuming way
        printWithThread("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> {
    rightExample().collect { value -> printWithThread("Collected $value") } 
}       

엇 그러면 여기서 의문이 든다. 이전에 Coroutine에서 활용했던 withContext는 쓸 수 없는 것인가?

일단 flow 블록 내부에서 withContext를 쓰는 두 개의 예시를 보자

 

 

1) Emit은 안 하고 flow 내부에서 withContext() 열기

fun myExample(): Flow<Int> = flow {
    withContext(Dispatchers.Default){printWithThread("Emit은 안해요")} 
    for (i in 1..3) {
        Thread.sleep(100) // pretend we are computing it in CPU-consuming way
        printWithThread("Emitting $i")
        emit(i) // emit next value
    }
}.flowOn(Dispatchers.IO)
​
fun main() = runBlocking<Unit> {
    withContext(Dispatchers.Default) {
        myExample().collect { value -> printWithThread("Collected $value") }
    }
}
​
// 출력 결과
[DefaultDispatcher-worker-3] Emit은 안해요
[DefaultDispatcher-worker-3] Emitting 1
[DefaultDispatcher-worker-1] Collected 1
[DefaultDispatcher-worker-3] Emitting 2
[DefaultDispatcher-worker-1] Collected 2
[DefaultDispatcher-worker-3] Emitting 3
[DefaultDispatcher-worker-1] Collected 3

 

2) withContext 내부에서 emit 하는 경우

fun wrongSimple(): Flow<Int> = flow {
    // The WRONG way to change context for CPU-consuming code in flow builder
    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
        }
    }
}
​
// main은 위 코드와 동일함
​
// 출력결과 : IllegalStateException 발생

 

4. 왜 와이 withContext 못쓰는데..? 👀

flow블록 내부에서 withContext를 왜 못쓰게 하는 것일까? 왜 그런지 궁금해서 GPT랑 좀 떠들어봤다.

• 나 : What kind of problem might occur when data is emitted on different threads?
  왜 emit 다른 스레드에서 하면 안 됨?

• GPT :
  1. Unexpected Ordering of Elements:emit 되는 데이터 순서 보장 안됨
  1. Mutable State and Shared Resources: 데이터가 공유되는 경우에 데이터 동기화 로직이 꼬일 수 있다
  2. Context Switching Overhead: 너무 당연한 이야기다. emit 할 때마다 context가 바뀌면 오버헤드가 일어날 수밖에 없을 것 같다.
  3. Unhandled Exceptions: 만약 emit 되는 데이터가 각각 다른 thread에서 방출되는데, 에러가 발생되는 경우에 exception을 처리하는 것도 쉽지 않은 일이 될 것이다
  4. Lost Context Information:매번 방출 때마다 context가 바뀌면 수집되는 쪽에서 emit 되는 데이터를 야무지게 처리하지 못함. (수집할 때 방출하는 쪽의 context를 사용하는 것을 유추해 볼 수 있었다)

 

그래서 결론은 flowOn을 활용해서 하나의 스레드에서 데이터를 일관되게 방출해야 한다는 것!

Consumer와 Producer의 Context는 다를 수 있다는 것!

끝!

해당 글은 Kotlin Document를 참고하여 작성했습니다.