[Kotlin Coroutine] Flow - 1

kotlinlang.org/docs/flow.html#sequences

Asynchronous Flow - Help | Kotlin

Asynchronous Flow

 

suspending 함수는 비동기적으로 하나의 값을 반환합니다. 비동기적으로 계산된 여러 값들을  어떻게 반환할 수 있을까요? Kotlin Flows는 여기서 등장합니다.

 

여러 값들을 표시하기

여러 값들은 Kotlin에서 collection을 이용해서 표현될 수 있습니다. 예를 들어서, simple() 함수는 세 개의 숫자들의 List를 반환하고 forEach를 이용해서 모든 숫자들을 출력합니다.

fun simple(): List<Int> = listOf(1, 2, 3)
 
fun main() {
    simple().forEach { value -> println(value) } 
}

결과

1
2
3

Sequence

만약 CPU를 사용해 blocking 코드(각 계산마다 100ms가 걸리죠)를 가지고 숫자들을 계산한다면, Sequence를 사용해서 숫자들을 나타낼 수 있습니다.

fun simple(): Sequence<Int> = sequence { // sequence builder
    for (i in 1..3) {
        Thread.sleep(100) // pretend we are computing it
        yield(i) // yield next value
    }
}

fun main() {
    simple().forEach { value -> println(value) } 
}

결과

1
2
3

코드 결과값은 같은 숫자들이지만, 하나씩 출력될 때마다 100ms가 걸립니다.

 

Suspending 함수

그러나, 이 계산은 코드를 수행하는 메인 쓰레드를 막습니다. 비동기 코드로 이 값들이 계산될 때, 우리는 simple 함수에 suspend 수정자를 표시하고, blocking 없이 코드는 돌아가고 리스트로 결과를 반환받을 수 있습니다.

import kotlinx.coroutines.*                 
                           
suspend fun simple(): List<Int> {
    delay(1000) // pretend we are doing something asynchronous here
    return listOf(1, 2, 3)
}

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

결과

1
2
3

이 코드는 1초를 기다리고 숫자들을 출력하죠.

 

Flows

List<Int> 결과 타입을 사용하는 것은, 오직 한 번에 모든 값들을 반환할 수 있다는 것을 의미합니다. 비동기적으로 계산되는 값들의 흐름을 나타내기 위해서, Flow<Int>타입을 비동기적으로 계산되는 값들을 위한 Sequence<Int> 타입을 사용하는 것처럼 사용할 수 있습니다.

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

              
fun simple(): Flow<Int> = flow { // flow builder
    for (i in 1..3) {
        delay(100) // pretend we are doing something useful here
        emit(i) // emit next value
    }
}

fun main() = runBlocking<Unit> {
    // Launch a concurrent coroutine to check if the main thread is blocked
    launch {
        for (k in 1..3) {
            println("I'm not blocked $k")
            delay(100)
        }
    }
    // Collect the flow
    simple().collect { value -> println(value) } 
}

결과

I'm not blocked 1
1
I'm not blocked 2
2
I'm not blocked 3
3

이 코드는 메인 쓰레드를 blocking 하지 않고 각 숫자들을 출력하기 전에 100ms를 기다립니다. 메인 쓰레드에서 돌고 있는 분리된 코루틴에서 매 100ms마다 "I'm not blocked"가 출력되는 것을 확인할 수 있습니다.

 

Flow와 이전 예제들 코드에서 다음과 같은 차이점들을 주목하세요.

 

• Flow 타입을 위한 빌더 함수가 flow를 호출합니다.
• flow{...} 빌더 블록 안의 코드는 일시중시할 수 있습니다.
• simple 함수에 suspend 수정자가 붙지 않습니다.
• emit 함수를 이용해서 flow에서 값들이 방출됩니다(emitted).
• collect 함수를 이용해서 flow에서 값들이 수집됩니다(collected).

 

Flow는 차갑습니다.

Flow는 sequence와 비슷하게 cold stream입니다. flow가 수집되기 전까지 flow 빌더 안의 코드는 돌아가지 않습니다. 아래의 예제를 통해 명확하게 알 수 있죠.

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

     
fun simple(): Flow<Int> = flow { 
    println("Flow started")
    for (i in 1..3) {
        delay(100)
        emit(i)
    }
}

fun main() = runBlocking<Unit> {
    println("Calling simple function...")
    val flow = simple()
    println("Calling collect...")
    flow.collect { value -> println(value) } 
    println("Calling collect again...")
    flow.collect { value -> println(value) } 
}

결과

Calling simple function...
Calling collect...
Flow started
1
2
3
Calling collect again...
Flow started
1
2
3

이것이 (flow를 반환하는)simple 함수가 suspend 수정자로 표시되지 않는 핵심 이유입니다. 그 자체로 simple() 호출은 빠르게 반환하고 아무것도 기다리지 않습니다. flow는 매 번 수집될 때마다 시작하고, 이것은 collect를 다시 호출할 때 "Flow started"를 보게되는 이유입니다.

 

Flow 취소 기초

Flow는 코루틴의 일반적인 협력 취소를 따르고 있습니다. 보통, flow collection은 flow가 (delay 처럼)취소 가능한 suspending 함수 안에서 일시중단될 때에 취소될 수 있습니다. 아래의 예제는 어떻게 flow가 withTimeoutOrNull 안에서 실행 중일 때 타임아웃으로 취소되고 코드 수행을 중지하는지를 보여줍니다.

 

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

          
fun simple(): Flow<Int> = flow { 
    for (i in 1..3) {
        delay(100)          
        println("Emitting $i")
        emit(i)
    }
}

fun main() = runBlocking<Unit> {
    withTimeoutOrNull(250) { // Timeout after 250ms 
        simple().collect { value -> println(value) } 
    }
    println("Done")
}

결과

Emitting 1
1
Emitting 2
2
Done

어떻게 simple 함수에서 숫자 두 개만이 flow에서 내보내게 되고 결과를 만드는지 주목하세요.

 

Flow 빌더들

이전 예제들에서 나온 flow{...} 빌더는 가장 기초적인 것입니다. flow의 더 쉬운 선언을 위하 다른 빌더들이 있습니다.

 

flowOf 빌더는 값의 고정된 집합을 내보내는 flow를 정의합니다.

다양한 collection들과 sequence들은 .asFlow() 확장 함수를 이용해서 flow로 변환될 수 있습니다.

 

1부터 3까지의 숫자를 출력하는 예제는 다음과 같이 작성할 수 있습니다.

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

fun main() = runBlocking<Unit> {
    // Convert an integer range to a flow
    (1..3).asFlow().collect { value -> println(value) } 
}

결과

1
2
3

 

중간 flow 연산자들

collection과 sequence와 마찬가지로 Flow는 연산자를 가지고 변형될 수 있습니다. 중간 연산자들은 upstream flow에 적용되어 downstream flow를 반환합니다. 이 연산자들은 flow와 마찬가지로 차갑습니다. 이러한 연산자를 호출하는 것은 자체로 suspending 함수는 아닙니다. 빠르게 동작하고, 새롭게 변형된 flow의 정의를 반환합니다.

 

기초 연산자들은 map, filter 같이 친숙한 이름들을 가지고 있습니다. 시퀀스와의 중요한 차이점은 이 연사자들 안의 코드 블록들은 suspending 함수들을 호출할 수 있다는 것입니다.

 

예를들어서, 입력 요청의 flow는 map 연산자로 결과를 매핑할 수 있습니다. 요청을 수행하는 것이 suspending 함수로 구현되는, 시간이 길게 걸리는 연산일 경우에도 말입니다. 

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

          
suspend fun performRequest(request: Int): String {
    delay(1000) // imitate long-running asynchronous work
    return "response $request"
}

fun main() = runBlocking<Unit> {
    (1..3).asFlow() // a flow of requests
        .map { request -> performRequest(request) }
        .collect { response -> println(response) }
}

다음 3줄을 만들고, 각 줄은 각 1초 후에 등장합니다.

response 1
response 2
response 3

변환 연산자

flow 변환 연산자들 중에서, 가장 일반적인 것은 transform이라 불립니다. map, filter 같이 간단한 변환을 모방하는데 사용될 수 있고, 더 복잡한 변환도 구현할 수 있습니다. transform 연산자를 이용해서, 임의의 횟수만큼 임의 값을 내보낼 수 있습니다.

 

예를 들어서, transform을 이용해 시간이 길게 걸리는 비동기 요청을 수행하기 전에 string을 내보낼 수 있고 응답을 뒤이어 보낼 수 있습니다.

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

suspend fun performRequest(request: Int): String {
    delay(1000) // imitate long-running asynchronous work
    return "response $request"
}

fun main() = runBlocking<Unit> {
    (1..3).asFlow() // a flow of requests
        .transform { request ->
            emit("Making request $request") 
            emit(performRequest(request)) 
        }
        .collect { response -> println(response) }
}

결과

Making request 1
response 1
Making request 2
response 2
Making request 3
response 3

크기 제한 연산자

take와 같이 크기 제한 중간 연산자는 해당하는 제한에 도달하면 flow의 수행을 취소합니다. 코루틴에서 취소는 항상  exception을 발생하면서 수행됩니다. 그래서 모든 리소스 관리 함수들(try {...} finally {...} 블록들)은 보통 취소하는 경우에 연산합니다.    

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

fun numbers(): Flow<Int> = flow {
    try {                          
        emit(1)
        emit(2) 
        println("This line will not execute")
        emit(3)    
    } finally {
        println("Finally in numbers")
    }
}

fun main() = runBlocking<Unit> {
    numbers() 
        .take(2) // take only the first two
        .collect { value -> println(value) }
}

결과

1
2
Finally in numbers

이 코드의 출력 결과는 numbers() 함수 안의 flow{...} body의 수행이 두 번째 숫자를 내보낸 이후에 중지된 것을 명확하게 보여줍니다. 

 

Terminal flow 연산자들

flow에서 Terminal 연산자들은 flow의 collection을 시작하는 suspending 함수들이다. collect 함수가 가장 대표적인 것이고, 더 쉬운 다른 terminal 연산자들이 있다.

 

• toList, toSet은 다양한 collection들로 변환시킨다.
• first 로 첫 값을 구하고, single로 값 하나를 내보낸다.
• reduce와 fold로 flow를 값으로 줄인다.

예시가 여기 있다.

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

fun main() = runBlocking<Unit> {

    val sum = (1..5).asFlow()
        .map { it * it } // squares of numbers from 1 to 5                           
        .reduce { a, b -> a + b } // sum them (terminal operator)
    println(sum)     
}

결과

55

flow는 순차적입니다.

flow의 각 개별 collection은 여러 flow들에서 연산하는 특정한 연사자들이 사용되지 않는한 순차적으로 수행됩니다. collection은 terminal 연사자를 호출하는 코루틴에서 바로 작업합니다. 보통 새로운 코루틴이 시작되지 않습니다. 방출된 각각의 값은 upsteam에서 downstream까지 모든 중간 연산자들에 의해서 처리되고 나서 후에 terminal 연산자로 전달됩니다.

 

짝수 정수들을 필터링하고 나서 문자열로 매핑하는 다음 예를 보세요.

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

fun main() = runBlocking<Unit> {

    (1..5).asFlow()
        .filter {
            println("Filter $it")
            it % 2 == 0              
        }              
        .map { 
            println("Map $it")
            "string $it"
        }.collect { 
            println("Collect $it")
        }                      
}

결과

Filter 1
Filter 2
Map 2
Collect string 2
Filter 3
Filter 4
Map 4
Collect string 4
Filter 5

Flow context

flow의 collection은 호출하는 코루틴의 context에서 항상 발생합니다. 예를 들어, simple flow가 있고, simple flow의 세부 구현사항과 관련없이 다음 코드는 이 코드의 작성자에 의해 지정된 context에서 실행합니다. 

withContext(context) {
    simple().collect { value ->
        println(value) // run in the specified context
    }
}

flow의 이러한 속성을 context perservation(보존)이라고 부릅니다.

보통, flow{...} 빌더 안의 코드는 해당하는 flow의 collector에 의해 제공된 context 안에서 실행합니다. 예를 들어, 호출된 쓰레드를 print하고 숫제 3개를 내보내는 simple 함수의 구현을 고려해보죠.

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

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
           
fun simple(): Flow<Int> = flow {
    log("Started simple flow")
    for (i in 1..3) {
        emit(i)
    }
}  

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

결과

[main @coroutine#1] Started simple flow
[main @coroutine#1] Collected 1
[main @coroutine#1] Collected 2
[main @coroutine#1] Collected 3

simple().collect 가 main thread에서 호출되었기 때문에, simple의 flow의 body는 main thread에서 호출되었습니다. 이것은 실행 context를 신경쓰지 않고 호출자를 차단하지 않는, 빠른 실행, 또는 비동기적인 코드를 위한 완벽한 기본값입니다.

 

잘못된 방출(emission) withContext

그러나, 장기간 CPU를 사용하는 코드는 Dispatchers.Default의 context에서 실행될 필요가 있고, UI를 업데이트하는 코드는 Dispatchers.Main의 context에서 실행되어야 합니다. 대개, withContext는 코틀린 코루틴을 사용하는 코드에서 context를 변경하는데 사용되는데요, 그러나 flow{...} 빌더 안의 코드는 context 보전 속성을 지켜야 하고, 다른 context에서 emit하는 건 허용하지 않습니다.

 

다음 코드를 시도해보죠.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
                      
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
        }
    }
}

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

결과. 다음과 같은 exception이 나옵니다.

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 ...

> flow 안에서 withContext 사용 못함.

 

flowOn 연산자

예외로 flow emission(방출)의 context를 바꾸는 데 사용하는 flowOn 함수가 있습니다. flow의 context를 바꾸는데 올바른 방법이 아래의 예에 보여집니다. 아래의 예는 해당하는 쓰레드가 어떻게 일하는지 보여주고 이름을 출력합니다.

 

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..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") 
    } 
}   

결과

[DefaultDispatcher-worker-1 @coroutine#2] Emitting 1
[main @coroutine#1] Collected 1
[DefaultDispatcher-worker-1 @coroutine#2] Emitting 2
[main @coroutine#1] Collected 2
[DefaultDispatcher-worker-1 @coroutine#2] Emitting 3
[main @coroutine#1] Collected 3

collection이 메인 쓰레드에서 동작하는 동안, flow{...}가 백그라운드 쓰레드에서 어떻게 일하는지를 알아두세요.

 

여기서 관찰해야 하는 다른 것은 flowOn 연산자가 flow의 기본 순차적인 특성을 바꾸었다는 것입니다. 현재 collection은 한 코루틴("coroutine#1)에서 동작하고, collect하는 코루틴과 함께 동시에 다른 쓰레드에서 돌고 있는 또 다른 코루틴("coroutine#2")에서 emission이 일어났습니다. flowOn 연사자는 context에서 CoroutineDispatcher를 바꿔야 할 때, upstream flow를 위해 또 다른 코루틴을 생성했습니다.

 

Buffering

다른 코루틴들안에서 한 flow의 다른 부분들을 실행하는 것은 flow를 수집하는 전체적인 시간의 관점에서는 유용할 수 있습니다. 특히 장시간 시행하는 비동기 연산을 포함해서 말이죠. 예를 들어서, simple flow의 emission이 느릴 때, 한 요소가 생성되는데 100ms가 걸리고, 수집하는 것도 느려서 한 요소를 처리하는데 300ms가 걸리는 경우를 고려해봅시다. 

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.*

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().collect { value -> 
            delay(300) // pretend we are processing it for 300 ms
            println(value) 
        } 
    }   
    println("Collected in $time ms")
}

결과

1
2
3
Collected in 1222 ms

buffer 연산자를 사용해서 순차적으로 실행하는 대신에, simple flow의 emitting 코드와 collectiong 코드를 동시에 실행할 수 있습니다.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.*

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")
}

결과

1
2
3
Collected in 1057 ms

같은 숫자들을 더 빠르게 생성하고, 처리 파이프라인을 효과적으로 생성해서, 첫 번째 숫자는 100ms만 기다리고 그 후에 각 숫자를 처리하는데에 300ms만 걸립니다.

 

Conflation / 합성

flow가 연산의 부분적인 결과들이나 연산 상태 업데이트의 일부 결과들을 나타낼 때, 모든 값을 처리할 필요는 없습니다. 대신에 가장 최근값만 처리하면 되죠. 이 경우에, conflate 연산자가 collector가 값들을 처리하는데 너무 오래 걸릴 때에 중간 값들을 건너뛰게 해줍니다. 이전 예제를 기반을 작성했습니다.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.*

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 756 ms

첫 번째 숫자가 아직도 처리 중일 동안에 두번째, 세번째 값이 이미 생산되었고요, 그래서 두 번째 값은 압축(conflated/결합)되어 가장 최근 값(세번째 값)이 collector로 전달되었습니다.

 

 최신 값 처리하기

Conflation은 emitter와 collector 둘 다 느릴 때 처리 속도를 빠르게 하는 하나의 방법입니다. 내보내진(emitted) 값들을 누락시키는 거죠. 다른 방법은 느린 collector를 취소하고 새 값을 내보내는 매 시간마다 collector를 재시작하는 겁니다. 이것은 xxx 연산자의 꼭 필수적인 로직을 수행하나 새 값을 받아 block 안의 코드는 취소하는  xxxLatest 연산자들의 일종입니다. 

 

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.*

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")
}

 

결과

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

 

collectLatest의 body는 300ms가 걸리나, 새 값들은 매 100ms마다 방출되니, block은 매 새 값마다 실행하나 마지막 값으로만 완료되는 걸 볼 수 있습니다.