[Kotlin][Android] Flow 사용하기

안녕하세요.

오늘은 코틀린에서 Flow를 사용해서 데이터를 관리하는 방법에 대해 알아보겠습니다.

 

2025.06.26 - [안드로이드 코틀린] - [Kotlin][Android] 코루틴(Coroutine) 사용하기

[Kotlin][Android] 코루틴(Coroutine) 사용하기

---

Flow란?

프로그래밍에서 데이터를 다루다 보면, 한 번에 끝나는 값이 아니라 시간에 따라 여러 값이 연속적으로 흘러가는 상황을 자주 마주하게 됩니다.

예를 들어, 센서의 측정값, 서버에서 순차적으로 받아오는 데이터, 사용자 입력 등은 모두 여러 값이 시간에 따라 흘러가는 구조입니다.

이러한 데이터의 흐름을 효과적으로 다루기 위해 코틀린에서는 Flow라는 표준 API를 제공합니다.

 

'값'과 '스트림'의 차이

일반 변수는 한 시점의 '값'만을 담습니다.

val age = 30

 

스트림(Stream)은 '값'들의 연속, 즉 시간에 따라 변화하는 데이터의 흐름을 다룹니다.

28, 29, 30, 31, ...

 

이처럼 스트림은 여러 값이 시간에 따라 차례로 흘러가는 개념입니다.

 

왜 Flow가 필요할까?

 

현대의 앱은 사용자 입력, 네트워크 응답, 데이터베이스 변화 등 언제 발생할지 모르는 비동기적 이벤트를 다루는 경우가 많습니다. 이런 이벤트를 효율적으로, 그리고 안전하게 처리하기 위해서는 데이터의 흐름을 비동기적으로 다룰 수 있는 구조가 필요합니다.

 

자바에서는 콜백, Observer 패턴, RxJava 등 다양한 방법으로 이를 처리했지만, 복잡하거나 코드가 장황해지는 단점이 있었습니다.

 

코틀린은 이러한 문제를 해결하고자 코루틴(Coroutine) 기반의 Flow를 도입했습니다.

Flow는 간결하고, 안전하며, 코틀린다운 방식으로 비동기 데이터 스트림을 처리할 수 있게 해줍니다.

 

---

Flow의 동작 구조

 

Flow는 생산자(Producer), 연산자(Operator), 소비자(Collector) 세 단계로 구성되어 있습니다.
이 구조를 이해하면 Flow의 작동 방식과 데이터가 어떻게 흘러가는지 명확하게 알 수 있습니다.

 

1. 생산자(Producer)

Flow를 생성하는 부분입니다. 데이터를 하나씩 emit() 함수로 방출합니다.

보통 flow { ... } 블록 안에서 사용합니다.

fun simpleFlow(): Flow<Int> = flow {
    for (i in 1..3) {
        emit(i) // 값을 하나씩 방출
    }
}

 

2. 연산자(Operator)

중간에서 데이터를 가공하는 역할을 합니다. 대표적으로 map, filter, take, onEach 등이 있습니다.

여러 연산자를 체이닝하여 복잡한 데이터 흐름도 간단하게 처리할 수 있습니다.

numberFlow()
    .map { it * 10 }      // 값을 10배로 변환
    .filter { it > 10 }   // 10보다 큰 값만 통과

 

3. 소비자(Collector)

 

Flow를 실제로 사용(수집)하는 부분입니다.

collect { ... } 블록 안에서 데이터를 받아 처리합니다.

이 collect가 호출되어야 Flow 전체가 실제로 동작합니다.

numberFlow()
    .map { it * 10 }
    .filter { it > 10 }
    .collect { value ->
        println("최종 수집: $value")
    }

 

 

결과

최종 수집: 20
최종 수집: 30

 

 

(1) 1 * 10 = 10  ->  filter에서 걸러짐

(2) 2 * 10 = 20  -> 출력

(3) 3 * 10 = 30 -> 출력

 

---

주요 연산자(Operators)와 예시

Flow는 다양한 연산자를 제공하여 데이터의 흐름을 쉽게 가공하고 제어할 수 있습니다. 대표적인 연산자와 그 예시는 다음과 같습니다.

 

map : 각 값을 변환

fun main() = runBlocking {
    simpleFlow()
        .map { it * 2 } // 각 값을 2배로 변환
        .collect { println(it) }
}

// [출력]
// 2
// 4
// 6

 

 

filter : 조건에 맞는 값만 통과

fun main() = runBlocking {
    simpleFlow()
        .filter { it % 2 == 1 } // 홀수만 통과
        .collect { println(it) }
}

// [출력]
// 1
// 3

 

take : 지정한 개수만 방출

fun main() = runBlocking {
    simpleFlow()
        .take(2) // 처음 두 개 값만 사용
        .collect { println(it) }
}

// [출력]
// 1
// 2

 

onEach : 각 값이 방출될 때마다 특정 동작 수행

fun main() = runBlocking {
    simpleFlow()
        .onEach { println("값 방출: $it") }
        .collect { println("수집: $it") }
}

// [출력]
// 값 방출: 1
// 수집: 1
// 값 방출: 2
// 수집: 2
// 값 방출: 3
// 수집: 3

 

catch : 예외 처리

fun errorFlow(): Flow<Int> = flow {
    emit(1)
    throw RuntimeException("에러 발생!")
}

fun main() = runBlocking {
    errorFlow()
        .catch { e -> println("예외 처리: ${e.message}") }
        .collect { println(it) }
}

// [출력]
// 1
// 예외 처리: 에러 발생!

 

---

Backpressure(역압)

Backpressure란, 데이터 생산 속도와 소비 속도가 다를 때 발생하는 문제를 의미합니다.

 

생산자가 데이터를 너무 빨리 방출하면 소비자가 따라가지 못해 데이터가 쌓이고, 메모리 부족 등 문제가 발생할 수 있습니다.

코틀린의 Flow는 기본적으로 suspend로 동작하기 때문에, 소비자가 데이터를 수집할 준비가 될 때까지 생산자가 기다립니다.
(즉, 기본적으로 Backpressure에 안전함).

fun fastProducer(): Flow<Int> = flow {
    repeat(1000) {
        emit(it)
        println("emit: $it")
    }
}

fun main() = runBlocking {
    fastProducer()
        .collect { value ->
            delay(100) // 소비자가 느림
            println("collect: $value")
        }
}

 

위 코드에서 생산자는 빠르게 데이터를 방출하지만, 소비자가 느리면 생산자가 자연스럽게 suspend되어 대기합니다.

이런 구조 덕분에 별도의 Backpressure 처리가 필요 없는 것이 코틀린 Flow의 큰 장점입니다.

 

---

Cold Stream vs Hot Stream

Cold Stream(콜드 스트림)

• 수집(collect)이 시작될 때마다 새로운 데이터 흐름이 시작됩니다.
• 여러 소비자가 동시에 collect하면 각각 독립적으로 데이터를 받음.
• 코틀린의 기본 Flow는 모두 Cold Stream입니다.

val coldFlow = flow {
    println("Flow 시작")
    emit(1)
    emit(2)
}

fun main() = runBlocking {
    coldFlow.collect { println("A: $it") }
    coldFlow.collect { println("B: $it") }
}
// Flow 시작
// A: 1
// A: 2
// Flow 시작
// B: 1
// B: 2

 

Hot Stream(핫 스트림)

• 데이터 자체가 계속 흘러가고 있고, 소비자가 collect할 때 그 시점의 데이터만 받음.
• 대표적인 예: StateFlow, SharedFlow, LiveData 등
• 여러 소비자가 동시에 collect해도 같은 데이터 스트림을 공유

StateFlow

• 항상 최신 상태(state)를 저장하는 Hot Stream
• 값이 변경될 때마다 구독자에게 최신 값 전달
• 기본값(initial value)이 필요하며, 항상 하나의 최신 값만 유지

val stateFlow = MutableStateFlow(0)

fun main() = runBlocking {
    launch {
        stateFlow.collect { println("수집: $it") }
    }
    delay(100)
    stateFlow.value = 1
    stateFlow.value = 2
    delay(100)
}

 

SharedFlow

• 여러 소비자에게 동일한 이벤트를 브로드캐스트하는 Hot Stream
• 버퍼를 설정해 최근 n개의 값을 새 구독자에게 전달할 수도 있음(Replay 기능)
• StateFlow와 달리 "상태" 개념 없이 이벤트를 전달하는 데 적합

val sharedFlow = MutableSharedFlow<Int>(replay = 2)

fun main() = runBlocking {
    launch {
        sharedFlow.collect { println("첫 번째 수집: $it") }
    }
    sharedFlow.emit(1)
    sharedFlow.emit(2)
    sharedFlow.emit(3)

    launch {
        sharedFlow.collect { println("두 번째 수집: $it") }
    }
    sharedFlow.emit(4)
}

 

구분	특징	대표 예시	용도/상황
Flow	Cold Stream, collect마다 새 흐름	flow { ... }	DB 쿼리, 네트워크, 일회성 데이터
StateFlow	Hot Stream, 최신 상태 유지	MutableStateFlow	UI 상태, 단일 상태 관리
SharedFlow	Hot Stream, 이벤트 브로드캐스트	MutableSharedFlow	알림, 여러 구독자 이벤트 전달