[Android] Flow FlatMap 어떻게 사용할까?

서론

코루틴 플로우를 자주 사용하면서 유독 flatMap을 잘 활용하지 않는다는 것을

깨달았습니다. flatMap 대신 익숙한 중간 연산자만 사용하다보니 이러면 정체될 것 같아서

flatMap도 적극 활용하기위해 포스팅합니다.

Zip과 Combine도 플로우를 조합, 결합하는데 사용하니 함께 다루겠습니다.

함께보시죠!

Zip

두 플로우를 결합해서 각요소를 조합해줍니다.

Zip은 동일한 위치에 있는 요소들을 하나로 묶어서 새로운 플로우를 생성합니다.

fun zip() = runBlocking {
    val flow1 = flowOf(1, 2, 3).onEach { delay(300) }
    val flow2 = flowOf("a", "b", "c").onEach { delay(400) }
    flow1.zip(flow2) { a, b ->
        "$a + $b"
    }.collect {
        println(it)
    }
}

실행값

1 + a
2 + b
3 + c

Combine

Zip과 마찬가지로 두 플로우를 조합하지만 Combine은 각 플로우가 방출 될 때마다

요소를 하나로 묶어서 값을 내보냅니다.

fun combine() = runBlocking {
    val flow1 = flowOf(1, 2, 3).onEach { delay(300) }
    val flow2 = flowOf("a", "b", "c").onEach { delay(400) }
    flow1.combine(flow2) { a, b ->
        "$a + $b"
    }.collect {
        println(it)
    }
}

실행값

1 + a
2 + a
2 + b
3 + b
3 + c

값이 방출되는 속도가 다른데 zip과 다르게 바로바로 조합해서 방출하는 것을 알 수 있습니다.

flattening 이란?

중첩된 플로우를 하나의 플로우로 평탄화 하는 것을 의미합니다.

그냥 map을 썼을땐 flow를 내부에서 또 처리해야합니다.

flatMap을 쓰면 평탄화 되서 바로 나옵니다.

FlatMapConcat

flatMap부터는 알기 쉽게 delay를 300으로 통일하겠습니다.

flatMapConcat은 순서대로 flattening을 하고 결과를 flow로 합칩니다.

println("______________flatMapConcat______________")
    val startTime = System.currentTimeMillis()
    val flow1 = flowOf(1, 2, 3).onEach { delay(300) }
    val flow2 = flowOf("a", "b", "c").onEach { delay(300) }
    flow1.flatMapConcat { a -> flow2.map { it + "$a"} }.collect{
        println("$it : ${System.currentTimeMillis() - startTime}")
    }

실행값

a1 : 626
b1 : 932
c1 : 1236
a2 : 1845
b2 : 2146
c2 : 2452
a3 : 3058
b3 : 3364
c3 : 3669

(300 + 300) * 6 약 3.6초가 걸리는 군요

1. 1 방출 300
2. a 조합 600 a1
3. b 조합 900 b1
4. c 조합 1200 c1
5. 2 방출 1500
6. a 조합 1800 a2
7. b 조합 2100 b2
8. c 조합 2400 c2
9. 3 방출 2700
10. a 조합 3000 a3
11. b 조합 3300 b3
12. c 조합 3600 c3

얼추 맞죠?

FlatMapMerge

flatMapMerge는 flatMapConcat과 다르게 병렬로 flattening을 진행합니다.

fun flatMapMerge() = runBlocking {
    val startTime = System.currentTimeMillis()
    val flow1 = flowOf(1, 2, 3).onEach { delay(300) }
    val flow2 = flowOf("a", "b", "c").onEach { delay(300) }
    flow1.flatMapMerge { a -> flow2.map { it + "$a"} }.collect{
        println("$it : ${System.currentTimeMillis() - startTime}")
    }
}

실행값

a1 : 635
b1 : 937
a2 : 937
a3 : 1242
c1 : 1242
b2 : 1242
b3 : 1549
c2 : 1549
c3 : 1854

merge같은 경우에는 순서 상관없이 먼저 처리 된 값 먼저 방출합니다.

값들을 병렬로 처리하기 때문에 각 플로우의 딜레이와 독립적으로 조합됩니다.

순서가 중요하지 않은 상황에서 활용하면 concat 과 비교하여 빠르게 처리가 가능합니다.

FlatMapLatest

flatMapLatest는 collectLatest와 유사합니다.

flattening이 진행 중일때 다음 요소의 flattening이 시작되면 취소하고 최신 요소만

처리합니다.

fun flatMapLatest() = runBlocking {
    val startTime = System.currentTimeMillis()
    val flow1 = flowOf(1, 2, 3).onEach { delay(300) }
    val flow2 = flowOf("a", "b", "c").onEach { delay(300) }
    flow1.flatMapLatest { a -> flow2.map { it + "$a"} }.collect{
        println("$it : ${System.currentTimeMillis() - startTime}")
    }
}

실행값

a1 : 630
a2 : 937
a3 : 1240
b3 : 1545
c3 : 1851

딜레이 도중에 새로운 것이 시작되면 취소되는 것을 볼 수 있습니다.

하고 싶은 말

flatMap 써보니까 그래도 느낌은 오는거 같네요

많이 공부해서 써먹어야 겠습니다.

참조

https://machine-woong.tistory.com/574