Coroutine flow

 

 코루틴의 flow 에 대해서 알아보겠습니다.

 

 일단 구글 문서부터 한번 살펴볼까요?

 

코루틴에서 흐름은 단일 값만 반환하는 정지 함수와 달리 여러 값을 순차적으로 내보낼 수 있는 유형입니다. 예를 들면 흐름을 사용하여 데이터베이스에서 실시간 업데이트를 수신할 수 있습니다.

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

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

 

 위와 같이 나와 있습니다. 여기에서 대략적으로 요약해보면 데이터의 스트림을 제어할 수 있다라고 나오는 것 같습니다.

 

 아직 뭔가 와닿지 않는 느낌인데 다른 곳을 한번 살펴보겠습니다.

 

플로우(Flows)

List<Int> 를 함수의 반환 타입으로 사용한다는 것은 결국 우리가 모든 연산을 수행 한 후 한번에 모든 값을 반환해야 함을 의미합니다. 비동기로 처리 될 값 들의 스트림을 나타내기 위해서 우리는 앞서 살펴본 동기적으로 처리되는 Senquence<Int> 타입에서 했던 것과 같이 Flow<Int> 타입을 사용할 수 있습니다.

 

 와 같이 나옵니다.

 

 간단하게 한마디로 정리하면 데이터의 흐름을 제어할 수 있다? 정도로 알면 될 것 같습니다.

 

 Coroutine flow에 대해서 검색해보면 

 

 그러면 예제를 통해서 한번 살펴볼까요?

 

 아래와 같이 기존에 list를 출력해보겠습니다.

 

fun main() {

    foo().forEach { value ->
        println(value)
    }

}

fun foo(): List<Int> = listOf(1, 2, 3)

 

 당연히 1,2,3이 출력됩니다.

 

 suspending functions으로 하면 아래와 같이 바꿀 수 있습니다.

 

suspend fun foo(): List<Int> {
    delay(1000) // pretend we are doing something asynchronous here
    return listOf(1, 2, 3)
}

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

 

 

 이 부분을 Flow로 한번 바꿔보겠습니다.

 

 flow {} 빌더를 이용하여 만들 수 있습니다.

 

 emit 이라는 친구가 있는데 이 친구가 값을 방출합니다.

 

fun foo(): 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> {

    // Collect the flow
    foo().collect {
        value -> println(value)
    }
}

 

   

 그리고, flow는 콜드스트림입니다.

 

 스트림에서 cold / hot 스트림 이 부분의 차이는

 

 cold stream의 경우, 구독했을 때 실행합니다.

 때문에, 모든 정보를 가져올 수 있습니다.

 

 아래와 같이 collect를 실행했을 때 collect로 구독하지 않으면 실행되지 않는 것을 볼 수 있습니다.

 

fun foo(): 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> {

    // Collect the flow
    foo()

}

 

 이와는 반대로 hot stream의 경우에는 

 

 subscribe 하지 않아도 스트림이 생성됩니다. 때문에 일정 시간이 지나고 구독하면, 앞의 데이터를 유실합니다.

 

 여기까지 flow에 대한 대략적인 설명이고, 어떤 것을 할 수 있는지 한번 보겠습니다.

 

 - 플로우 빌더 Flow Builders

 

 그러면 flow{} 라는 형태가 아닌 다른 형태로도 만들 수 있을까요?

 

 아래와 같이 asFlow를 이용해서 만들어줄 수 있습니다.

 

fun test(): Flow<Int> {


    val arr = ArrayList<Int>()
    arr.add(0)
    arr.add(1)
    arr.add(2)

    return arr.asFlow()

}

fun main() = runBlocking {

    test().collect{ value ->
        println(value)
    }


}

 

 

map

 

 

Flow 에서 suspend function을 이용해서 데이터를 가공하고 싶을 때 map을 사용해주면 됩니다.

 

아래와 같이 사용할 수 있습니다.

 

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

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

}

 

filter

 

아래와 같이 필터링 해줄 수 있습니다.

 

fun main() = runBlocking {
    (1..5).asFlow()
        .filter {
            it % 2 == 0
        }.collect {
            println("Collect $it")
        }
}

 

reduce

 

마지막 값만 반환해주는 것도 가능합니다.

 

fun main() = runBlocking {
    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)
}

 

 

take

 

하나만 가져오고 나머지는 가져오지 않는 방식입니다.

 

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 {
    numbers()
        .take(2) // take only the first two
        .collect { value -> println(value) }
}

 

combine

 

combine 같은 경우는 flow가 방출될 때 마다 연산을 처리해줄 수 있습니다.

 

fun main() = runBlocking {
//    val nums = (1..3).asFlow().onEach { delay(300) } // numbers 1..3 every 300 ms
//    val strs = flowOf("one", "two", "three").onEach { delay(400) } // strings every 400 ms
//    val startTime = System.currentTimeMillis() // remember the start time
//    nums.zip(strs) { a, b -> "$a -> $b" } // compose a single string with "zip"
//        .collect { value -> // collect and print
//            println("$value at ${System.currentTimeMillis() - startTime} ms from start")
//        }


    val nums = (1..3).asFlow().onEach { delay(300) } // numbers 1..3 every 300 ms
    val strs = flowOf("one", "two", "three").onEach { delay(400) } // strings every 400 ms
    val startTime = System.currentTimeMillis() // remember the start time
    nums.combine(strs) { a, b -> "$a -> $b" } // compose a single string with "combine"
        .collect { value -> // collect and print
            println("$value at ${System.currentTimeMillis() - startTime} ms from start")
        }

}

  

 

 

 

 

 

 

 

- 참조

 

https://developer88.tistory.com/86

Hot Observable 과 Cold Observable은 무엇인가요?

 

https://gift123.tistory.com/64

안드로이드 개발 (33) Coroutine Flow on Android

 

https://developer.android.com/kotlin/flow

Android의 Kotlin 흐름  |  Android 개발자  |  Android Developers

 

https://myungpyo.medium.com/%EC%BD%94%EB%A3%A8%ED%8B%B4-%EA%B3%B5%EC%8B%9D-%EA%B0%80%EC%9D%B4%EB%93%9C-%EC%9E%90%EC%84%B8%ED%9E%88-%EC%9D%BD%EA%B8%B0-part-9-a-d0082d9f3b89

코루틴 공식 가이드 자세히 읽기 — Part 9-A