[코틀린 코루틴] 플로우란 무엇인가?

플로우(flow)는 비동기적으로 계산해야 할 값의 스트림을 나타낸다.

flow 인터페이스 자체는 떠다니는 원소들은 모으는 역할을 하며, 플로우의 끝에 도달할 때까지 각 값을 처리하는 걸 의미한다(flow의 collect는 컬렉션의 foreach와 비슷).

Flow의 유일한 멤버함수는 collect이며, Iterable과 Sequence와 비슷하다.

interface Flow<out T> {
    suspend fun collect(collector: FlowCollector<T>)
}

interface Iterable<out T> {
    operator fun iterator(): Iterator<T>
}

interface Sequence<out T> {
    operator fun iterator(): Iterator<T>
}

 

 

플로우와 값들을 나타내는 다른 방법들의 비교

 

예를 들어, 여러 개의 값을 반환하는 함수가 필요하다고 가정했을때 컬렉션을 사용하는 것보다, 원소를 하나씩 계산할 수 있는 sequence을 사용하는 것이 적절하다.

// 컬렉션 사용
fun getList(): List<String> = List(3) {
   Thread.sleep(1000)
   "User$it"
}

fun main() {
   val list = getList()
   println("Function started")
   list.forEach { println(it) }
}
//(3초 후)
//Function started
//User0
//User1
//User2

// 시퀀스 사용
fun getSequence(): Sequence<String> = sequence {
   repeat(3) {
       Thread.sleep(1000)
       yield("User$it")
   }
}

fun main() {
   val list = getSequence()
   println("Function started")
   list.forEach { println(it) }
}
//Function started
//(1초 후)
//User0
//(1초 후)
//User1
//(1초 후)
//User2

시퀀스는 동기적 데이터 생산을 지원하는 구조로, 코루틴의 중단 지점(suspension point) 개념을 포함하지 않기 때문에 동기적으로 데이터를 반환하는데 최적화되어 있는 yield와 같은 SequenceScope의 리시버에서 호출되는 함수 외에는 중단 함수는 사용할 수 없다.

중단 함수를 사용할 수 없다.

 

시퀀스는 스레드 블로킹을 하기 때문에, 다음 예제와 같은 결과가 나오게 된다.

fun getSequence(): Sequence<String> = sequence {
    repeat(3) {
        Thread.sleep(1000)
        yield("User$it")
    }
}

suspend fun main() {
    withContext(newSingleThreadContext("main")) {
        launch {
            repeat(3) {
                delay(100)
                println("Processing on coroutine")
            }
        }

        val list = getSequence()
        list.forEach { println(it) }
    }
}
//(1초 후)
//User0
//(1초 후)
//User1
//(1초 후)
//User2
//(0.1초 후)
//Processing on coroutine
//(0.1초 후)
//Processing on coroutine
//(0.1초 후)
//Processing on coroutine

 

플로우를 사용하면 코루틴이 연산을 수행하는 데 필요한 기능을 전부 사용할 수 있다.

fun getFlow(): Flow<String> = flow {
    repeat(3) {
        delay(1000)
        emit("User$it")
    }
}

suspend fun main() {
    withContext(newSingleThreadContext("main")) {
        launch {
            repeat(3) {
                delay(100)
                println("Processing on coroutine")
            }
        }

        val list = getFlow()
        list.collect { println(it) }
    }
}
//(0.1초 후)
//Processing on coroutine
//(0.1초 후)
//Processing on coroutine
//(0.1초 후)
//Processing on coroutine
//(1 - 0.1 * 3 = 0.7초 후)
//User0
//(1초 후)
//User1
//(1초 후)
//User2

 

 

플로우의 특징

 

• 플로우는 스레드 블로킹하지 않고 코루틴을 중단시킨다.
• 플로우는 코루틴 컨텍스트를 활용하고 예외를 처리하는 등의 코루틴 기능을 제공한다.
• 플로우 빌더는 중단 함수가 아니며 어떠한 스코프도 필요로 하지 않는다.
• 플로우의 최종 연산은 중단 가능하며, 연산이 실행될 때 부모 코루틴과의 관계가 정립된다.

fun usersFlow(): Flow<String> = flow {
    repeat(3) {
        delay(1000)
        val ctx = currentCoroutineContext()
        val name = ctx[CoroutineName]?.name
        emit("User$it in $name")
    }
}

suspend fun main() {
    val users = usersFlow()

    withContext(CoroutineName("Name")) {
        val job = launch {
            // collect는 중단함수이다.
            users.collect { println(it) }
        }

        launch {
            delay(2100)
            println("I got enough")
            job.cancel()
        }
    }
}
//(1초 후)
//User0 in Name
//(1초 후)
//User1 in Name
//(0.1초 후)
//I got enough

 

 

플로우 명명법

 

• 플로우 빌더, 다른 객체에서의 변환 도는 헬퍼 함수로 시작
• 플로우의 마지막은 최종 연산이라 불리며, 주로 collect가 된다.
• 시작 연산과 최종 연산 사이에 플로우를 변경하는 중간 연산을 가질 수 있다.

suspend fun main() {
    flow { emit("Message 1") }                          // 플로우 빌더
        .onEach { println(it) }                         // 중간 연산
        .onStart { println("Do something before") }     // 중간 연산
        .onCompletion { println("Do something after") } // 중간 연산
        .catch { emit("Error") }                        // 중간 연산
        .collect { println("Collected $it") }           // 최종 연산
}