Flow — 개념, Kotlin 코루틴의 콜드 및 핫 스트림

Kotlin에서 Flow란?

Flow는 kotlinx.coroutines.flow 패키지의 타입으로, 콜드 비동기 데이터 스트림을 나타냅니다. 본질적으로 Flow는 emit() 함수를 통해 값을 방출하고 성공 또는 예외로 종료되는 코루틴 시퀀스입니다. 스트림 수집은 서스펜드 함수인 터미널 연산자 collect()를 통해 수행됩니다.

콜드 시맨틱스

콜드 스트림은 flow 빌더 내부의 코드가 각 구독자마다 새로 실행됨을 의미합니다. RxJava의 Observable.fromIterable도 유사하게 동작합니다: 새 구독자는 처음부터 모든 값을 받습니다. Flow에서는 서스펜드 함수 collect를 통해 구현되며, 데이터 수집 전체 기간 동안 코루틴을 블로킹합니다.

Flow 빌더

Kotlin은 Flow를 생성하는 여러 방법을 제공합니다: flow { } — emit()을 사용한 기본 구성, flowOf(vararg values) — 고정 값 세트용, .asFlow() — 컬렉션 및 Sequence용 확장. 모든 빌더는 콜드입니다 — 데이터는 터미널 연산자가 호출될 때만 생성됩니다.

콜드 스트림과 핫 스트림

콜드 스트림과 핫 스트림의 구분은 리액티브 프로그래밍의 핵심 개념입니다. 콜드 스트림 (Flow, Observable)은 구독 시 데이터 생성을 시작합니다. 핫 스트림 (Channel, SharedFlow)은 독립적으로 데이터를 방출합니다 — 구독자는 시퀀스의 시작 없이 구독 이후에 발생하는 것만 받습니다.

SharedFlow는 여러 구독자를 가질 수 있고 replay가 구성된 경우 최근 값을 재생할 수 있는 핫 Flow입니다. SharedFlow는 이벤트(일회성 알림)에 적합합니다. StateFlow는 고정된 상태 값을 가진 변형으로, 새 구독자를 위해 최신 값을 캐시합니다.

ChannelFlow는 내부적으로 Channel을 사용하여 Flow와 Channel의 속성을 결합합니다. capacity를 통한 버퍼링과 백프레셔를 지원합니다. ChannelFlow는 콜백 API를 리액티브 스트림으로 변환할 때 유용하며, 여러 코루틴에서 값이 방출됩니다.

콜드와 핫 간 변환

콜드 Flow를 핫 SharedFlow로 변환하려면 shareIn(scope, started, replay) 연산자를 사용합니다. started 매개변수는 시작 시점을 제어합니다: SharingStarted.WhileSubscribed() — 구독자가 있는 동안 활성, Lazily — 첫 구독자에서 시작, Eagerly — 즉시 시작. 역변환 — 핫에서 콜드: StateFlow.asFlow()는 콜드 Flow를 반환하며 collect 시 StateFlow의 현재 값을 방출합니다. 이는 테스트에 편리합니다.

Flow 연산자

Flow는 코루틴 내에서 서스펜드 함수로 작동하는 풍부한 연산자 세트를 제공합니다. 연산자는 상태가 없으며 새 Flow를 반환합니다 — 원본 스트림은 변경되지 않습니다. 이를 통해 부작용 없이 안전한 변환 체인을 구축할 수 있습니다.

map 연산자는 각 스트림 값을 비동기 또는 동기 변환을 통해 변환합니다. filter는 조건을 충족하는 값만 통과시킵니다. catch는 터미널 연산자 전에 예외를 포착하고 스트림 복구를 허용합니다. flatMapLatest는 새 값이 도착하면 이전 방출을 취소합니다 — Rx의 switchMap과 유사합니다.

Flow의 debounce 연산자는 지정된 타임아웃까지 값 게시를 지연시킵니다. 이 시간 동안 새 값이 도착하면 타이머가 재설정됩니다. Android에서 debounce는 검색에 사용됩니다: 300-400ms 일시 중지 후에만 요청이 전송되어 API 호출이 3-5배 감소합니다.

터미널 연산자

collect() 외에도 Flow는 다른 터미널 연산자를 지원합니다: toList()는 모든 값을 목록으로 수집 — 테스트에 유용, first()는 첫 번째 요소를 반환하고 스트림을 취소, single()은 정확히 하나의 요소를 기대합니다. fold(initial)는 전달된 함수를 통해 값을 누적합니다. 모든 터미널 연산자는 서스펜드 함수이며 코루틴 또는 다른 서스펜드 함수 내에서 호출해야 합니다.

Flow 코드 예제

첫 번째 예제 — map 연산자를 통한 변환으로 숫자를 생성하는 기본 Flow:

val numberFlow = flow {
    for (i in 1..5) {
        delay(500)
        emit(i)
    }
}

scope.launch {
    numberFlow
        .map { "숫자: $it" }
        .collect { value ->
            println(value)
        }
}

두 번째 예제 — catch를 통한 필터링 및 오류 처리가 포함된 스트림 변환:

flow {
    emit("data1")
    emit("data2")
    throw RuntimeException("network error")
}
    .catch { e ->
        emit("fallback_data")
    }
    .collect { value ->
        println(value)
    }

세 번째 예제 — Jetpack Compose에서 반응형 UI를 위한 ViewModel의 StateFlow 사용:

class SearchViewModel : ViewModel() {
    private val _query = MutableStateFlow("")
    val results: StateFlow<List<Result>> = _query
        .debounce(300)
        .flatMapLatest { query ->
            repository.search(query)
        }
        .catch { emit(emptyList()) }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), emptyList())

    fun onQueryChanged(query: String) {
        _query.value = query
    }
}

StateFlow와 SharedFlow

StateFlow는 단일 현재 값을 가진 핫 Flow입니다. 최신 값을 캐시하고 새 구독자에게 즉시 전달합니다. StateFlow는 상태를 위한 관찰 가능한 컨테이너이며 equals 비교를 지원합니다 — 새 값이 현재 값과 일치하면 방출이 발생하지 않습니다. Jetpack Compose는 collectAsState()를 통해 StateFlow를 사용합니다.

SharedFlow는 필수 초기 값 없이 더 유연한 핫 Flow입니다. SharedFlow는 replay(새 구독자용 값 수), extraBufferCapacity(replay를 초과하는 버퍼), onBufferOverflow(오버플로우 시 전략)를 통해 구성됩니다. SharedFlow는 일회성 이벤트에 이상적입니다: 네비게이션, Snackbar, 분석.

Android 아키텍처에서 Flow는 Google에 의해 기본 데이터 소스(계층: Repository → UseCase → ViewModel)로 권장됩니다. LiveData는 유연성에서 Flow보다 떨어집니다: Flow는 코루틴, 연산자, 백프레셔를 지원하고 UI 계층 외부에서 작동합니다. LiveData에서 Flow로의 마이그레이션은 현대 Android 프로젝트에서 표준 관행입니다.

ViewModel에서 Flow를 사용할 때 올바른 타입을 선택하는 것이 중요합니다. StateFlow는 화면 회전을 견뎌야 하는 UI 상태에 이상적입니다. SharedFlow는 재처리가 허용되지 않는 이벤트에 적합합니다 — 예: 네비게이션. lifecycleScope에서 collect()를 사용한 Flow는 실행 컨텍스트에 대한 최대 제어를 제공하지만 화면 이탈 시 수동 취소가 필요합니다.

Flow 테스트는 kotlinx-coroutines-test를 통해 수행됩니다. 라이브러리는 TestDispatcher를 제공합니다 — 지연(delay)을 가속화하고 코루틴 실행 순서를 제어할 수 있는 가상 시간입니다. TestScope.runTest { }는 Flow 테스트를 위한 격리된 환경을 만듭니다. toList() 연산자는 스트림이 올바른 데이터 시퀀스를 방출했는지 확인하기 위해 타임아웃과 함께 모든 flow 값을 수집하는 테스트에서 자주 사용됩니다.

Flow는 Room(Android DB 라이브러리)과 잘 통합됩니다: DAO 메서드는 Flow<List<Entity>>를 반환할 수 있습니다. Room은 테이블 변경 시 자동으로 새 값을 방출합니다 — 수동 트리거 없이 UI가 업데이트됩니다. 이는 내부적으로 callbackFlow와 함께 Flow를 사용하는 InvalidationTracker를 통해 구현됩니다. 이 접근 방식은 LiveData의 필요성을 제거하고 데이터 계층을 완전히 코루틴 지향적으로 만듭니다. Jetpack Compose는 collectAsState()를 통해 StateFlow를 구독하고 데이터가 변경된 컴포넌트만 다시 그립니다 — 이는 LiveData 지향 아키텍처로는 달성할 수 없는 성능을 제공합니다. DataStore(SharedPreferences 대체)도 Flow<Preferences>를 반환하여 수동 업데이트 트리거 없이 애플리케이션 설정의 반응형 읽기를 제공합니다.

Flow는 추가 라이브러리 없이 JVM에서 kotlinx-coroutines-core를 통해 프로세스 간 통신을 지원합니다. 예를 들어, Ktor 서버 애플리케이션에서 Flow는 수신 WebSocket 메시지 스트림을 나타낼 수 있습니다. 각 메시지는 스트림으로 방출되고 연산자를 통한 필터링과 집계를 거쳐 결과가 클라이언트로 전송됩니다. 이 접근 방식은 Kotlin 프로젝트에서 Reactor 또는 RxJava와 같은 리액티브 라이브러리를 대체합니다.

기존 RxJava 코드와의 Flow 호환성은 kotlinx-coroutines-rx3 모듈에 의해 제공됩니다. 확장 함수 Flow.asObservable()은 Flow를 RxJava 3의 Observable로 변환합니다. 역변환 — CompletableSource.asFlow(), Observable.asFlow(). 이는 RxJava에서 코루틴으로의 마이그레이션을 단순화합니다: 일부 계층을 RxJava에 남겨두고 프로젝트를 단계적으로 다시 작성할 수 있습니다. 변환 시 콜드/핫 시맨틱스의 차이를 고려해야 합니다: Observable은 콜드와 핫 모두 가능하지만 Flow는 일반 Flow의 경우 항상 콜드, SharedFlow의 경우 핫입니다.

Flow 오류 처리 및 테스트

Flow의 오류 처리에는 특징이 있습니다: 터미널 연산자 전에 flow 빌더 내에서 예외가 발생하면 catch로 전파됩니다. 빌더 후 연산자에서 예외가 발생하면 해당 연산자 후의 catch가 포착합니다. retryWhen은 조건부로 구독 재시도를 허용합니다: 네트워크 오류 시 최대 3회 재시도, 그러나 CancellationException에서는 재시도하지 않습니다. Flow는 상태를 저장하지 않으므로 상태 종속 오류를 제거합니다 — 이는 Subject가 내부 상태를 저장하는 Observable과 비교하여 디버깅을 단순화합니다.

kotlinx-coroutines-test를 사용한 Flow 테스트는 지연을 시뮬레이션하기 위해 TestDispatcher를 사용합니다. Turbine은 Flow 테스트를 위한 인기 있는 커뮤니티 라이브러리입니다: test { }는 Flow를 시작하고, awaitItem()은 다음 값을 기다리며, awaitComplete()는 완료를 기다립니다. Turbine은 기본 타임아웃을 추가하여 테스트 중단을 방지합니다. StateFlow 테스트를 위해서는 시간순 값 검증과 함께 .testIn(scope)를 사용하세요.