flatMapConcat, flatMapMerge, flatMapLatest차이

리액티브 스트림 라이브러리를 쓰면 수많은 intermediate연산자들이 있다. 그러한 연산자를 얼마나 많이 알고 활용해서 쓸 수 있는지가 바로 리액티브스티림 사용에 핵심이라고 볼 수 있다. 이번 포스팅은 그 중, 3가지 연산자를 알아보고자 한다.

• flatMapConcat
• flatMapMerge
• flatMapLatest

연산자를 이해하는데 있어 각 단어들을 알고가면 이해가 훨씬 더 쉽다. 위 메소스들의 차이를 이해하기 위한 핵심 단어인 'flat', 'map', 'concat', 단어들을 이해하는것으로 먼저 시작해보자.

1. 단어의 이해

1.1. flat

이는 flatten을 의미하며 '평평하게 만든다'는걸 의미한다. 리액티브 프로그래밍에선 배열이나 하나의 스트림을 의미하며 즉, 여러개의 배열이나 스트림을 평평하게 만든다는 뜻이다. 아래와 같이 말이다.

[
    [123],
    [456],
    [789]
] 
=> [1,2,3,4,5,6,7,8,9]

1.2. concat

이는 concatenate라는 뜻으로, 두 개 이상의 배열을 조합하여 하나의 배열 또는 스트림을 만드는 것을 의미한다. 아래와 같이 말이다.

[1,2,3] 
[4,5,6] 
[7,8,9]
=> [1,2,3,4,5,6,7,8,9]

1.3. map

이는 각 배열 내의 요소들을 반환하여 새로운 요소로 만들어내는 것을 의미한다. 아래와 같이 말이다.

val list = listOf(1,2,3) 
val newList = list.map { "$it" + "hello" }
=> ["1hello", "2hello", "3hello"]

단어정리를 다 했으니 이제 3메소드들 차이를 알아보자.

2. 메소드의 이해

2.1. flatMapConcat

flatMapConcat은 평탄화 관련 중간 연산자에 있어 가장 기본이 되는 메서드이다. 아래 소스코드와 마블 다이어그램을 통해 설명을 해볼까 한다.

fun main() {
    CoroutineScope(Dispatchers.IO).launch {
        flowOf("A","B","C")
            .flatMapConcat { productNewFlow(it) }
            .collect { Log.i("flatMapConcatTest", it) }
    }
}

fun productNewFlow(element: String) =
    flowOf(1,2,3)
        .onEach { delay(1000L) }
        .map { "$it + $element" }
        
/* 결과값
1 + A
2 + A
3 + A
1 + B
2 + B
3 + B
1 + C
2 + C
3 + C
*/

마블 차트 다이어그램을 그려보면 다음과 같다.

즉, 하위 데이터 생산부의 delay가 어떤지 상관 없이, 상위 데이터 생산자는 방출할 값을 버퍼에 저장해두고 있는걸 알 수 있다. 그리고 하위 데이터 생산자의 작업이 모두 끝났을 때, 데이터 처리가 차근차근 일어나는걸 볼 수 있다. 즉, 동기적으로 데이터 소비가 일어난다는 것이다.

[핵심]
flatMapConcat은 동기적으로 스트림을 결합한다.

2.1. flatMapMerge

마찬가지로 소스코드와 마블 다이어그램을 통해 설명을 해볼까 한다.

fun main() {
    CoroutineScope(Dispatchers.IO).launch {
        flowOf("A","B","C")
            .flatMapMerge { productNewFlow(it) }
            .collect { Log.i("flatMapMergeTest", it) }
    }
}

fun productNewFlow(element: String) =
    flowOf(1,2,3)
        .onEach { delay(1000L) }
        .map { "$it + $element" }
/*결과값
1 + A
1 + B
1 + C
2 + A
2 + B
2 + C
3 + A
3 + C
3 + B        
*/

위 코드는 flatMapConcat과 조금은 다르다는걸 확인할 수 있다. 바로 하위 데이터 생산부쪽의 데이터 처리가 완료되지 않았음에도(=delay가 걸려있음에도) 상위 데이터 생산부에서 데이터를 방출하면 그에 맞게 데이터가 소비된다는걸 알 수 있다. 즉, 비동기적으로 스트림이 소비된다는 의미이다.

[핵심]
flatMapMerge는 비동기적으로 스트림을 결합한다.

2.3. flatMapLatest

이는 위에서 알아보았던 flatMapConcat과 flatMapMerge와는 쓰임새가 조금은 다르다. 위에서 보았던 함수의 경우, 데이터 생산부에서는 delay를 걸어주지 않았다. 즉, 다음의 공식이 성립했었다.

[flatMapConcat과 flatMapMerge의 공통점]
상위 스트림의 데이터 생산 속도 > 하위 스트림의 데이터 생산 속도

반면, flatMapLatest의 경우는 아래의 경우 사용할 수 있다.

[flatMapLatest?]
상위 스트림의 데이터 생산 속도 < 하위 스트림의 데이터 생산 속도

아래 코드를 보자.

fun main() {
    CoroutineScope(Dispatchers.IO).launch {
        flowOf("A","B","C")
            .onEach { delay(1200L) }
            .flatMapLatest { productNewFlow(it) }
            .collect { Log.i("flatMapLatestTest", it) }
    }
}

fun productNewFlow(element: String) =
    flowOf(1,2,3)
        .onEach { delay(1000L) }
        .map { "$it + $element" }
        
/*결과값
1 + A
1 + B
1 + C
2 + C
3 + C
*/

위 코드를 보면 알다시피, 상위 스트림의 데이터 생산속도가 더 길다. 상위 스트림의 데이터 생산 속도는 1.2초마다 생산이 되며, 하위 스트림의 데이터 생산 속도는 1초이다.

즉, flatMapLatest메서드의 'Latest'의 이름에 걸맞게 가장 최신으로 방출되는 데이터 스트림만 반영한다는걸 알 수 있다.(=대기중이던 데이터 처리는 반영되지 않음)

이를 마블 차트 다이어그램으로 그려보면 다음과 같다.

마블 차트 다이어그램을 계속 보면 금방 이해가 갈거리 생각한다. 이는 상위 데이터 스트림에서 최신 데이터가 방출되었을 때, 하위 스트림의 데이터 처리는 cancel되게 되며, 상위 스트림과 sync를 맞춘 새로운 stream을 생산하게 된다.

그리고 상위 스트림에서 추가적인 데이터 생산이 없다면, 기존에 예정되어 있던 하위 스트림의 데이터 처리가 이뤄지게 된다.

[핵심]
flatMapLatest는 하위 스트림 데이터 처리 여부를 상관하지 않는다. 오로지 상위 스트림의 최신 데이터가 방출되면 기존에 진행되던 하위 스트림의 데이터 처리는 cancel시키고 상위 스트림에 맞는 데이터 처리를 진행한다.

3. 마치며

리액티브 프로그래밍에선 수많은 intermediate연산자들이 있다. zip, combine, map, 등등.. 이러한 연산자들을 영어단어처럼 하나하나씩 알고 있으면 보일러 플레이트 코드를 작성하지 않고, 좀 더 깔끔한 프로그래밍이 가능하다 생각한다.