[Kotlin Coroutine] Flow 공식 문서 번역 및 정리 (1)

 

---

• 불필요한 코드나 잘못 작성된 내용에 대한 지적은 언제나 환영합니다. 👍
• Kotlin Coroutine 공식 문서 내의 Asynchronous Flow 페이지를 공부하며 이해하기 쉽도록 한글 문맥으로 더 자연스럽게 번역 및 정리한 글입니다.

---

 

suspending function은 하나의 값을 비동기적으로 반환하지만, 비동기적으로 계산된 여러 값을 반환하는 방법은 무엇일까요? 이것이 Kotlin Flow가 필요한 이유입니다.

 

Representing multiple values

 

Kotlin에서는 컬렉션을 사용하여 여러 값을 나타낼 수 있습니다. 예를 들어, 세 개의 숫자를 포함하는 List를 반환하는 간단한 함수를 작성하고 forEach를 사용하여 모두 출력할 수 있습니다:

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

결과

1
2
3

 

Sequences

 

만약 CPU 소비가 많은 블로킹 코드를 사용하여 숫자를 계산한다면(각 계산에 100ms가 소요됨), Sequence를 사용하여 숫자를 나타낼 수 있습니다. 이 코드는 같은 숫자를 출력하지만, 각 숫자를 출력하기 전에 100ms를 기다립니다.

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

 

 

Suspending functions

 

그러나 이 계산은 코드를 실행하는 메인 스레드를 차단합니다. 이러한 값이 비동기 코드에 의해 계산될 때, simple() 함수에 suspend 한정자를 표시하여 차단하지 않고 작업을 수행하고 결과를 리스트로 반환할 수 있습니다. 이 코드는 1초를 기다린 후 숫자를 출력합니다.

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

 

 

Flows

 

List<Int> 타입을 결과로 사용하면 한 번에 모든 값을 반환할 수 있습니다. 비동기적으로 계산되는 값의 스트림을 나타내기 위해 Flow<Int> 유형을 사용할 수 있습니다. 이는 동기적으로 계산되는 방식인 Sequence<Int> 유형을 사용하는 것과 비슷합니다.

 

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

 

이 코드는 메인 스레드를 차단하지 않고 각 숫자를 출력하기 전에 100ms를 기다립니다. 이것은 별도의 코루틴에서 "I'm not blocked"를 100ms마다 출력되는 것을 통하여 확인할 수 있습니다. 이 별도의 코루틴은 메인 스레드에서 실행됩니다.
이전 예제와 달리 다음과 같은 차이점이 있습니다.

 

• Flow 유형의 빌더 함수는 flow로 불립니다.
• flow { ... } 빌더 블록 내부의 코드에 suspend function을 호출할 수 있습니다.
• simple 함수는 더 이상 suspend 한정자로 표시되지 않습니다.
• emit 함수를 사용하여 flow에 value를 발행(emit)합니다.
• collect 함수를 사용하여 flow에서 value를 수집(collect)합니다.

 

✏️ simple 함수의 flow { ... } 블록 내부의 delay를 Thread.sleep으로 바꾸면, 메인 스레드가 차단되는 것을 확인할 수 있습니다.

 

---

 

Flows are cold

 

Flow는 시퀀스와 유사한 cold stream입니다. flow builder 내의 코드는 flow가 수집(collect)될 때까지 실행되지 않습니다. 다음 예제로 확인해 보겠습니다.

 

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

 

이것이 simple 함수가 (flow를 반환) suspend 한정자로 표시되어 있지 않은 주요한 이유입니다. simple 함수 자체는 빠르게 반환되며 어떤 것도 기다리지 않습니다. Flow는 매번 수집될 때마다 새롭게 다시 시작되기 때문에 collect를 다시 호출할 때마다, "Flow started"가 출력되는 것입니다.

 

---

 

Flow cancellation basics

 

플로우는 코루틴의 일반적인 협력 취소에 따라 실행됩니다. 일반적으로, 플로우 수집은 취소 가능한 suspending function(예 : delay)에서 일시 중단되면 취소될 수 있습니다. 다음 예제는 withTimeoutOrNull 블록에서 실행될 때 타임아웃으로 플로우가 취소되고 코드 실행이 중단되는 방법을 보여줍니다.

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

 

다음과 같이 simple 함수의 flow에서는 두 개의 숫자만이 emit 되어 출력됩니다.

Emitting 1
1
Emitting 2
2
Done

Flow 취소에 관한 자세한 내용은 이후에 나올 Flow cancellation checks 섹션에서 다룹니다.

 

---

 

Flow builders

 

이전 예제에서 사용된 flow { ... } 빌더는 가장 기본적인 빌더입니다. 플로우를 선언할 수 있는 다른 빌더도 있습니다.

 

• flowOf 빌더는 고정된 값 집합을 방출하는 흐름을 정의합니다.
• 여러 컬렉션과 시퀀스는 .asFlow() 확장 함수(extension function)를 사용하여 플로우로 변환될 수 있습니다.

 

예를 들어, 숫자 1에서 3까지 출력하는 코드를 다음과 같이 다시 작성할 수 있습니다.

// Convert an integer range to a flow
(1..3).asFlow().collect { value -> println(value) }

 

---

 

Intermediate flow operators (중간 연산자)

 

컬렉션 및 시퀀스가 ​​변환되는 방식과 유사하게 연산자를 사용하여 흐름을 변환할 수 있습니다. 중간 연산자는 업스트림 플로우에 적용되고, 다운스트림 플로우를 반환합니다. 이러한 연산자는 플로우와 마찬가지로 cold 성격입니다. 이러한 연산자를 호출하는 것은 suspending function이 아니므로, 빠르게 작동하여 새롭게 변환된 플로우를 정의하고 반환합니다.

 

기본 연산자들은 map과 filter와 같은 익숙한 이름을 가지고 있습니다. 시퀀스와 중요한 차이점은 이러한 연산자 내부의 코드 블록이 suspending function을 호출할 수 있다는 것입니다.

예를 들어 아래와 같이 Request가 suspending function으로 구현된 장기적인 비동기 작업인 경우에도, map 연산자를 사용하여 Request의 결과를 맵핑하고 각 요청의 결과를 얻을 수 있습니다.

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

예시 코드는 각 라인마다 1초씩 딜레이 되면서 총 3줄의 메시지를 출력합니다.

response 1
response 2
response 3

 

Transform operator (변형 연산자)

 

transform 연산자는 가장 일반적인 플로우 변형 연산자 중 하나입니다. map이나 filter 같은 간단한 변형부터 더 복잡한 변형까지 구현할 수 있습니다. transform 연산자를 사용하여 임의의 값을 임의의 횟수로 발행(emit)할 수 있습니다.

예를 들어, transform을 사용하여 긴 비동기 요청을 수행하기 전에 문자열을 먼저 발행하고, 이후에 response를 받으면 해당 response를 사용할 수 있습니다.

 

(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

 

Size-limiting operators (크기 제한 연산자)

 

take와 같은 크기 제한 중간 연산자는 해당 한계치에 도달하면 플로우의 실행을 취소합니다. 코루틴에서 취소는 exception을 throw 하여 수행되며, 이렇게 하면 모든 리소스 관리 함수 (예: try { ... } finally { ... } 블록)가 취소되는 경우에도 정상적으로 작동합니다.

 

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) // 먼저 emit 되는 2개를 취함
        .collect { value -> println(value) }
}

이 코드의 출력 결과를 보면, numbers() 함수 안의 flow { ... } 블록의 실행이 두 번째 숫자까지 발행한 뒤에 멈추는 것을 확인할 수 있습니다.

1
2
Finally in numbers

 

---

[Kotlin Coroutine] Flow 공식 문서 번역 및 정리 (2) 에서 계속됩니다.

---