[XX캠퍼스] 10.Kotlin Coroutines & Flow ( Flow 컨텍스트 )

플로우는 코루틴 컨텍스트에서

플로우는 현재 코루틴 컨텍스트에서 호출 됩니다.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
           
fun simple(): Flow<Int> = flow {
    log("flow를 시작합니다.")
    for (i in 1..10) {
        emit(i)
    }
}  

fun main() = runBlocking<Unit> {
    launch(Dispatchers.IO) {
        simple()
            .collect { value -> log("${value} 를 받음.") } 
    }
}

실행결과 
[DefaultDispatcher-worker-1 @coroutine#2] flow를 시작합니다.
[DefaultDispatcher-worker-1 @coroutine#2] 1를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 2를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 3를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 4를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 5를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 6를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 7를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 8를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 9를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 10를 받음.

Dispatcher.IO에서 호출이 된다. 

 

 

 

다른 컨텍스트로 옮겨갈 수 없는 플로우

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
           
fun simple(): Flow<Int> = flow {
    withContext(Dispatchers.Default) {  // 플로우 내에서는 컨텍스트를 바꿀수 없다 . 
        for (i in 1..10) {
            delay(100L)
            emit(i)
        }
    }
}  

fun main() = runBlocking<Unit> {
    launch(Dispatchers.IO) {
        simple()
            .collect { value -> log("{$value} 를 받음.") } 
    }
}

플로우 내에서는 컨텍스트를 바꿀수 없다 . 

Exception in thread "main" java.lang.IllegalStateException: Flow invariant is violated:

 

 

---

flowOn 연산자

flowOn 연산자를 통해 컨텍스트를 올바르게 바꿀 수 있습니다.

 

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
           
fun simple(): Flow<Int> = flow {
    for (i in 1..10) {
        delay(100L)
        log("값 ${i}를 emit합니다.")
        emit(i)
    } // 업스트림  // 얘만 Dispatchers.Default 
}.flowOn(Dispatchers.Default) // 위치 
 .map{
 	it *2 // 다운스트림
 }

fun main() = runBlocking<Unit> {
    simple().collect { value ->  // 다운스트림
        log("${value}를 받음.")
    } 
}

실행결과 
[DefaultDispatcher-worker-1 @coroutine#2] 값 1를 emit합니다.
[main @coroutine#1] 1를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 값 2를 emit합니다.
[main @coroutine#1] 2를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 값 3를 emit합니다.
[main @coroutine#1] 3를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 값 4를 emit합니다.
[main @coroutine#1] 4를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 값 5를 emit합니다.
[main @coroutine#1] 5를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 값 6를 emit합니다.
[main @coroutine#1] 6를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 값 7를 emit합니다.
[main @coroutine#1] 7를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 값 8를 emit합니다.
[main @coroutine#1] 8를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 값 9를 emit합니다.
[main @coroutine#1] 9를 받음.
[DefaultDispatcher-worker-1 @coroutine#2] 값 10를 emit합니다.
[main @coroutine#1] 10를 받음.

 

.flowOn(Dispatchers.Default) 기준으로 위에있으면 업스트림,  아래있으면 다운스트림

map도 Dispatchers.Default 로 하고 싶다면 map을 .flowOn(Dispatchers.Default) 위로 옮겨준다. 

 

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
           
fun simple(): Flow<Int> = flow {
    for (i in 1..10) {
        delay(100L)
        log("값 ${i}를 emit합니다.")
        emit(i)
    }
} // 업스트림 Dispatchers.IO 
.flowOn(Dispatchers.IO)
.map{
	it*2
    // 업스트림 Dispatchers.Default
}.flowOn(Dispatchers.Default)
.flowOn(Dispatchers.IO)  // 실수로 하나 더 붙인경우는  위에 Default 가 적용 


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