FLOW란?

flow는 kotlin에서 제공하는 비동기 스트림을 나타내는 개념이다.Flow는 대량의 데이터를 처리하거나, 시간이 지남에 따라 변하는 데이터를 처리해야 할 때 유용하게 사용한다.
flow는 연속적으로 데이터를 생성하며, 이 데이터는 비동기적으로 처리될수 있다
Flow는 Cold Stream의 한 종류여서, 데이를 수집하는 코드가 데이터를 요청하기전까지는 아무런 데이터를 생성하지 않습니다.

주요메소드

• flow {}: Flow를 생성하는 빌더입니다. 이 블록 내에서 emit() 함수를 사용하여 값을 발행할 수 있습니다.
• emit(): Flow에 값을 발행합니다. emit() 함수는 flow {} 블록 내에서만 사용할 수 있습니다.
• collect(): Flow를 수집합니다. Flow가 발행하는 각 값을 처리하는 람다를 인자로 받습니다.
• asFlow(): 컬렉션을 Flow로 변환합니다
• filter(): Flow의 각 요소 중 특정 조건을 만족하는 요소만을 선택합니다. 선택 조건을 인자로 받는 람다를 인자로 받습니다.
• collectLatest(): Flow를 수집합니다. 새로운 값이 emit되었을 때 이전에 수집하던 작업을 취소하고 새로운 값의 수집을 시작합니다. 각 값에 대해 처리하는 람다를 인자로 받습니다

예제

Flow1.kt

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

fun main() {

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

}

1,2,3을 반환하는 함수를 출력하면 1,2,3이 나온다

Flow2.kt

import kotlinx.coroutines.*

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

}

1초 대기 한 후에 1,2,3을 출력한다

Flow3.kt

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
        println("delay(100)")
        emit(i) // emit next value
    }
}

fun main() = runBlocking<Unit> {

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

1초대기하고 1부터 3까지 정수가 순차적으로 emit(발생)한다.
이를 collect로 수집하고 출력한다

delay(100)
1
delay(100)
2
delay(100)
3

Flow4.kt

// Flow 는 Cold Stream 이다

// Cold Stream -> 수도꼭지 -> 내가 틀면 물이 나온다
// Hot Stream -> 냇가(계곡) -> 그냥 물이 흐르고 있다

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

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

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

collecct를 사용하기전까진 안나오는걸 알수있다

Calling foo...
Calling collect...
Flow started
1
2
3

​
Flow5.kt

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

// Flow builders

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

1,2,3콜랙션을 .asFlow()를 사용해서 flow로 변환했다.

1
2
3

​
Flow6.kt

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

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

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

}

.map을 사용하여 performRequest()함수에 인자를 넣고 collect로 값을 출력한다.

1
4
9

​
Flow7.kt

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


fun main() = runBlocking<Unit> {

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

}

.filter를 사용하여 특정 조건에 만족하는 요소만 선택한다.

Collect 2
Collect 4

​
Flow8.kt

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

// collectLatest
//
fun main() = runBlocking<Unit> {

    flow {
        emit(1)
        delay(50)
        emit(2)
    }.collectLatest { value ->
        println("Collecting $value")
        // delay(30) // Emulate work
        delay(80) // Emulate work
        println("$value collected")
    }

}

1이 emit되고 50초 후에 2가 emit 된다. 근데 collectLatest를 사용하면 새로운 값이 emit되었을 때 이전에 수집하던 작업을 취소하고 새로운 값을 나옴으로 아래와 같이 결과가 나온다

Collecting 1
Collecting 2
2 collected