[Coroutine Flow] flatMapMerge 을 사용해 flow 변환 동시 처리하기

flatMapMerge는 무슨 역할을 하는가?

flatMapConcat과 flatMapLatest는 flow에서 발행된 데이터를 변환할 때 발행된 순서대로 순차적으로 변환한다. 반대로 flatMapMerge는 변환을 병렬로 수행한다.

 

대부분의 연산이 flatMapConcat이나 flatMapLatest를 이용한 순차 처리에 해당하지만 들어오는 데이터들을 동시에 수집한 후 수집한 값들이 가능한 빨리 방출 될 수 있도록 병렬로 처리되어야 할 때가 있다. 예를 들어 비용 처리를 위해 수십개의 지출 데이터를 취합하여 합치는 작업을 할 경우 굳이 순차적으로 처리하지 않고 병렬로 처리되는 것이 빠를 것이다. flatMapMerge는 이러한 병렬 연산을 지원하기 위해 만들어진 연산자이다.

 

 

flatMapConcat과 flatMapMerge의 차이

아래와 같이 1과 5를 발행하는 flow에 대해 +1, +2, +3 변환을 수행하면서 flatMapConcat을 사용해보자. 그러면 1에 대한  변환값인 2, 3, 4 와 5에 대한 변환 값인 6, 7, 8이 순서대로 나온다.

fun collectWithFlatMapConcat() {
    val flow = flow<Int> {
        emit(1)
        emit(5)
    }

    viewModelScope.launch {
        flow.flatMapConcat { intValue ->
            flow {
                emit(intValue + 1)
                delay(1000)
                emit(intValue + 2)
                emit(intValue + 3)
            }
        }.collect {
            println("printed value >> $it")
        }
    }
}

그림1. flatMapConcat 결과

 

 위의 코드와 같이 변환 연산자 중간에 delay가 있다면 연산이 순차적으로 처리되는데는 시간이 오래 걸릴 것이다. 만약 순차적으로 처리되지 않아도되는 데이터라면 데이터들을 동시에 수집한 후 수집한 값들이 가능한 빨리 방출 될 수 있도록 병렬로 처리 하는 것이 효율적일 것이다. flatMapMerge는 이러한 병렬 연산을 지원하기 위해 만들어진 연산자이다.

 

위의 flatMapConcat을 flatMapMerge로 바꾸어 어떻게 결과가 나오는지 보도록 하자.

fun collectWithFlatMapMerge() {
    val flow = flow<Int> {
        emit(1)
        emit(5)
    }

    viewModelScope.launch {
        flow.flatMapMerge { intValue ->
            flow {
                emit(intValue + 1)
                delay(1000)
                emit(intValue + 2)
                emit(intValue + 3)
            }
        }.collect {
            println("printed value >> $it")
        }
    }
}

flatMapMerge를 사용하면 둘째 발행값인 5는 첫째 발행값인 1에 대한 변환이 완료되는 것을 기다리지 않고 변환을 시작한다. 따라서 다음의 순서로 변환이 처리된다.

 

1. 1이 발행되어 변환된 2가 발행된 후 1에 대한 변환은 1초간 쉰다.

2. 5가 발행되어 변환된 6이 발행되며, 5에 대한 변환은 1초간 쉰다.

3. 그 사이 1은 1초를 모두 쉬고 3과 4를 마저 발행한다.

4. 그 후 5는 1초를 모두 쉬고 7과 8을 발행한다. 

 

그림2. flatMapMerge

 

flatMapMerge의 의의

순차 처리는 리소스를 최대한 활용하기가 어려워 시간이 오래 걸린다. 특히 요즘과 같이 멀티코어 환경이 보편화된 시대에는 순차 처리가 필요하지 않은 곳에서 병렬 처리를 하면 수배에서 수십배까지 연산 속도가 빨라질 수 있다.

 

순서가 중요하지 않은 변환에서는 flatMapConcat이나 flatMapLastest 대신 flatMapMerge를 사용하면 연산 속도를 수십배 빠르게 만들 수 있다.