Coroutine Flow 란?

HEETAE HEO·2022년 9월 13일

Flow 란?

순차적으로 값을 배출해서, 정상적으로 완료하거나에러를 던지는 비동기 데이터 스트림 입니다. 일시중단할 수 있는 함수는(Suspending function) 비동기적으로 하나의 값을 반환합니다. 하지만 비동기적으로 계산되어진 값을 여러개를 반환할 때 Flow를 사용합니다.

코루틴에서 데이터 스트림을 구현하기 위해서의 구성요소는 다음과 같습니다.

Producer, Intermediary(선택사항), Consumer

Flow Builder

flow {}를 사용하는 것이 가장 기본적인 flow Builder입니다. emit() 이외에 asFlow()를 통하여 Collection 및 Sequence를 Flow로 변환할 수 있습니다.

(1..10).asFlow().collect { value -> 
	println(value)
}

Producer

생산자에서는 데이터를 발행하기 위하여 flow{} 코루틴 block(builder)을 생성한 후 내부에서 emit()을 통하여 데이터를 생성합니다. 또한 flow{} block은 suspend 함수이므로 delay 호출할 수 있습니다.

fun flowSomething(): Flow<Int> = flow {
	repeat(10) {
    	emit(it)
        delay(100L)
    }
}

Intermediary

생산자에서 데이터를 발행을 하였다면 중간 연산자는 생성된 데이터를 수정할 수 있습니다.
Kotlin의 Collection의 함수와 같이 대표적으로 map(데이터를 원하는 형태로 변환), filter(데이터 필터링), onEach(데이터를 변경 후 수행한 결과를 반환) 등이 있습니다.

flowSomethings().filter {
	it % 2 == 0 // 짝수 필터링
}


flowSomethings().map {
	it * 2 // 값을 2배씩
}

Consumer

생산자에서 데이터를 발행하고, 중간 연산자에서 데이터를 가공하였다면 소비자에서는 collect()를 이용하여 전달된 데이터를 소비할 수 있습니다.

fun main() = runBlocking {
    flowSomething().map {
        it * 2
    }.collect { value ->
        println(value)
    }
}
// 0 2 4 6 8 .. 18

2. flow 더 알아보기

Flow는 코루틴을 기반으로 빌드되며 여러 값을 제공할 수 있습니다. Flow는 비동기식으로 계산할 수 있는 데이터 스트림의 개념입니다. 내보낸 값은 동일한 유형이어야 합니다. 예를 들어 Flow는 정수 값을 내보내는 흐름입니다.

Flow는 값 시퀀스를 생성하는 Iterator와 비슷하지만 정지 함수를 사용하여 값을 비동기적으로 생성하고 사용합니다. 예를 들어 Flow는 기본 스레드를 차단하지 않고 다음 값을 생성할 네트워크 요청을 안전하게 만들 수 있습니다.

1. 간단한 flow 사용

fun flowSomething(): Flow<Int> = flow {
    repeat(10) {
        emit(it) //0 1 2 3 ... 9
        delay(10L)
    }
}

fun main() = runBlocking {
    flowSomething().collect { value ->
        println(value)
    }
}

결과

0
1
2
3  
...
 
9

flow 플로우 빌더 함수를 이용해서 코드블록을 구성하고 emit을 호출해서 스트림에 데이터를 흘려 보냅니다. 플로우는 콜드 스트림이기 때문에 요청 측에서 collect를 호출해야 값을 발생하기 시작합니다.

콜드 스트림 - 요청이 있는 경우에 보통 1:1로 값을 전달하기 시작.
핫 스트림 - 0개 이상의 상대를 향해 지속적으로 값을 전달.

2. 플로우 취소

fun flowSomething(): Flow<Int> = flow {
    repeat(10) {
        emit(it) //0 1 2 3 ... 9
        delay(100L)
    }
}

fun main() = runBlocking<Unit> {
    val result = withTimeoutOrNull(500L) {
        flowSomething().collect { value ->
            println(value)
        }
        true
    } ?: false
    if (!result) {
        println("취소되었습니다.")
    }
}

withTimeoutOrNull을 이용해 취소할 수 있습니다. withTimeoutOrNull(500L)라고 하게 되면 500ms 가 되었을 때 withTimeoutOrNull이 null을 리턴하기 때문에 결과적으로 false를 반환하게 됩니다.

3. 플로우 빌더 flowOf

flow 이외에도 몇가지 flowOf, asFlow등의 플로우 빌더가 있습니다. 먼저 flowOf를 살펴봅니다. flowOf는 여러 값을 인자로 전달해 플로우를 만듭니다.

fun main() = runBlocking<Unit> {
    flowOf(1, 2, 3, 4, 5).collect { value ->
        println(value)
    }
}

결과
  
1
2
3
4
5
4. 플로우 빌더 asFlow

asFlow는 컬렉션이나 시퀀스를 전달해 플로우를 만들 수 있습니다.

fun main() = runBlocking<Unit> {
    listOf(1, 2, 3, 4, 5).asFlow().collect { value ->
        println(value)
    }
    (6..10).asFlow().collect {
        println(it)
    }
}

3. flow 연산하기

1. flow와 map

플로우에서 map 연산을 통해 데이터를 가공할 수 있습니다.

fun flowSomething(): Flow<Int> = flow {
    repeat(10) {
        emit(it)
        delay(10L)
    }
}

fun main() = runBlocking {
    flowSomething().map {
        "$it $it"
    }.collect { value ->
        println(value)
    }
}

2. flow와 filter

fun main() = runBlocking<Unit> {
    (1..20).asFlow().filter {
        (it % 2) == 0 //짝수만 필터링
    }.collect {
        println(it)
    }
}

결과
  
2
4
6
  
...
  
20

3. filterNot

만약 홀수만 남기고 싶을 때 코드를 홀수에 맞게 수정할 수 도 있습니다. 하지만 짝수를 출력하는 코드를 그대로 두고 filterNot을 사용할 수도 있습니다.

fun main() = runBlocking<Unit> {
    (1..20).asFlow().filterNot {
        (it % 2) == 0
    }.collect {
        println(it)
    }
}

4. transform 연산자

위의 중간 연산자(map, filter)은 요소마다 1개의 변환밖에 하지 못하지만 변환 연산자(transform)은 예시처럼 emit()을 추가하여 요소마다 여러개의 변환이 가능하게 해줍니다.

suspend fun someCalc(i: Int): Int {
    delay(10L)
    return i * 2
}

fun main() = runBlocking<Unit> {
    (1..20).asFlow().transform {
        emit(it)
        emit(someCalc(it))
    }.collect {
        println(it)
    }
}

5. take 연산자

take 연산자는 몇개의 수행 결과만 취합니다.

suspend fun someCalc(i: Int): Int {
    delay(10L)
    return i * 2
}

fun main() = runBlocking<Unit> {
    (1..20).asFlow().transform {
        emit(it)
        emit(someCalc(it))
    }.take(5)
    .collect {
        println(it)
    }
}