KO
FrançaisDeutschEnglishEspañolPortuguêsTiếng Việt한국어简体中文日本語
홈 / 프로그래밍 언어 / Kotlin

Kotlin Flow: 비동기 데이터 스트림

Cris The Dev · 2026. 10. 13. · 9 분 읽기 · 103

1. 개요

suspend 함수는 하나의 값만 반환합니다. 여러 값을 시간이 지남에 따라 받아야 할 때, 예를 들어 새로운 주문, 결제 상태 또는 주식 가격이라면, 데이터 스트림이 필요합니다. Kotlin Flow가 바로 그 역할을 합니다.

이 글은 Flow가 무엇이고 연산자를 사용해서 데이터를 변환하는 방법을 설명합니다. 그 후 flowOn으로 스레드를 전환하고, 오류를 처리하고, StateFlow와 SharedFlow를 사용하는 방법, 그리고 Flow를 테스트하는 방법을 배울 것입니다.

2. 설치

Flow는 kotlinx-coroutines 라이브러리에 있습니다. 이 글은 Maven 프로젝트가 이미 kotlin-maven-plugin으로 구성되어 있다고 가정합니다. 다음 의존성만 추가하면 됩니다:

<dependencies>
    <dependency>
        <groupId>org.jetbrains.kotlinx</groupId>
        <artifactId>kotlinx-coroutines-core</artifactId>
        <version>1.10.2</version>
    </dependency>

    <!-- Test -->
    <dependency>
        <groupId>org.jetbrains.kotlinx</groupId>
        <artifactId>kotlinx-coroutines-test</artifactId>
        <version>1.10.2</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.junit.jupiter</groupId>
        <artifactId>junit-jupiter</artifactId>
        <version>5.11.4</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.assertj</groupId>
        <artifactId>assertj-core</artifactId>
        <version>3.27.3</version>
        <scope>test</scope>
    </dependency>
</dependencies>

Spring Boot 3.x를 사용하는 경우 coroutines 버전이 BOM을 통해 관리되므로 <version> 태그를 생략할 수 있습니다.

3. 첫 번째 Flow

3.1. Flow 생성 및 수집

Flow를 생성하는 가장 간단한 방법은 flow { } 빌더를 사용하고 각 값에 대해 emit()을 호출하는 것입니다:

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking

data class Order(val id: Int, val amount: Long)

fun orderStream(): Flow<Order> = flow {
    println("주문 발행 시작")
    for (id in 1..3) {
        delay(100)                      // 시스템에서 데이터를 기다리는 것을 시뮬레이션
        emit(Order(id, id * 50_000L))
    }
}

fun main() = runBlocking {
    val orders = orderStream()
    println("Flow를 생성했습니다")
    orders.collect { println("받음: $it") }
}

설명:

  • orderStream()은 suspend 함수가 아닙니다. 단지 데이터 스트림에 대한 설명만 반환합니다.
  • flow { } 내부에서 delay()와 같은 suspend 함수를 호출할 수 있습니다.
  • collect는 terminal operator입니다. 이를 호출할 때만 flow { } 내의 코드가 실행됩니다.

결과:

Flow를 생성했습니다
주문 발행 시작
받음: Order(id=1, amount=50000)
받음: Order(id=2, amount=100000)
받음: Order(id=3, amount=150000)

“Flow를 생성했습니다” 줄이 “주문 발행 시작”보다 먼저 나타납니다. 이는 Flow를 생성하는 것이 아무것도 트리거하지 않음을 증명합니다.

3.2. Flow는 “cold”입니다

flow { }로 생성된 Flow는 cold stream입니다. 각 collect 호출마다 전체 코드 블록이 처음부터 다시 실행됩니다:

fun main() = runBlocking {
    val orders = orderStream()
    orders.collect { }
    orders.collect { }
}
주문 발행 시작
주문 발행 시작

참고: flow { } 블록이 API를 호출하거나 데이터베이스에 쓰는 경우 각 collect마다 다시 호출됩니다. 이것이 많은 “왜 API가 두 번 호출되었나?” 버그의 원인입니다. 여러 collector가 하나의 소스를 공유하려면 섹션 6을 참고하세요.

4. 변환 연산자

4.1. Intermediate operator

Flow는 collection처럼 map, filter, take, onEach 등의 익숙한 연산자들을 가지고 있습니다. 이러한 연산자들은 intermediate입니다. 이들은 새로운 Flow를 반환하며 원래 Flow처럼 lazy입니다.

import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.map

fun main() = runBlocking {
    orderStream()
        .filter { it.amount >= 100_000 }
        .map { "주문 #${it.id}: ${it.amount} VND" }
        .collect { println(it) }
}
주문 발행 시작
주문 #2: 100000 VND
주문 #3: 150000 VND

각 요소는 전체 filter → map → collect 체인을 통과한 후 다음 요소가 발행됩니다. Flow는 기본적으로 같은 coroutine 내에서 순차적으로 실행됩니다.

collection의 map과 달리 Flow의 map의 lambda는 suspend입니다. 따라서 비동기 함수를 바로 내부에서 호출할 수 있습니다:

suspend fun fetchCustomerName(orderId: Int): String {
    delay(50)
    return "고객 $orderId"
}

orderStream()
    .map { fetchCustomerName(it.id) }
    .collect { println(it) }

4.2. 다른 Terminal operator

collect 외에도 다른 terminal operator를 사용하여 직접 결과를 얻을 수 있습니다:

import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.fold
import kotlinx.coroutines.flow.toList

fun main() = runBlocking {
    val all = orderStream().toList()
    val firstBig = orderStream().first { it.amount > 60_000 }
    val revenue = orderStream().fold(0L) { acc, order -> acc + order.amount }

    println("주문 수: ${all.size}, 가장 큰 주문: #${firstBig.id}, 수익: $revenue")
}
...
주문 수: 3, 가장 큰 주문: #2, 수익: 300000

first { }은 적합한 요소를 찾으면 즉시 Flow를 취소합니다. 따라서 그 collect에서 주문 #3은 절대 발행되지 않습니다.

5. flowOn으로 스레드 전환

기본적으로 flow { } 내의 코드는 collect를 호출하는 coroutine의 context에서 실행됩니다. 파일이나 데이터베이스를 읽는 등 데이터 발행 부분을 다른 스레드로 옮기려면 flowOn을 사용합니다:

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.flowOn

data class Customer(val name: String)

fun loadCustomers(): Flow<Customer> = flow {
    println("데이터를 읽는 중 ${Thread.currentThread().name}")
    listOf("An", "Bình").forEach { emit(Customer(it)) }
}.flowOn(Dispatchers.IO)

fun main() = runBlocking {
    loadCustomers().collect {
        println("처리 중 ${it.name} ${Thread.currentThread().name}")
    }
}
데이터를 읽는 중 DefaultDispatcher-worker-1
처리 중 An main
처리 중 Bình main

flowOn은 upstream에만 영향을 줍니다. 즉, 그 앞의 연산자들입니다. collect 부분은 여전히 호출자의 context에서 실행됩니다. 이 원칙을 context preservation이라고 합니다.

참고: flow { } 내에서 emit()을 감싸기 위해 withContext를 사용하지 마세요. 다음 코드는 IllegalStateException: Flow invariant is violated를 던집니다:

flow {
    withContext(Dispatchers.IO) { emit(1) } // 잘못됨
}

항상 flowOn을 사용하세요. 여러 coroutine에서 발행해야 한다면 channelFlow { }를 사용하세요.

6. 오류 처리

6.1. catch와 onCompletion

Flow의 Exception은 일반 코드처럼 collect로 전파됩니다. catch 연산자를 사용하면 오류를 차단하고 대신 값을 발행할 수 있습니다:

import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.onCompletion

fun paymentStatus(): Flow<String> = flow {
    emit("PENDING")
    throw IllegalStateException("결제 게이트웨이 타임아웃")
}

fun main() = runBlocking {
    paymentStatus()
        .catch { e -> emit("FAILED: ${e.message}") }
        .onCompletion { cause -> println("완료됨, cause=$cause") }
        .collect { println(it) }
}
PENDING
FAILED: 결제 게이트웨이 타임아웃
완료됨, cause=null

설명:

  • catch는 upstream의 오류만 포착합니다. collect의 lambda에서 던진 Exception은 포착되지 않습니다.
  • onCompletion은 catch 뒤에 있습니다. 오류가 이전 단계에서 처리되었으므로 cause는 null입니다. 두 연산자의 위치를 바꾸면 cause가 exception을 포함하는 것을 보게 됩니다.

6.2. retry

네트워크 끊김 같은 일시적 오류의 경우 retry를 사용할 수 있습니다. 이 연산자는 처음부터 다시 upstream Flow를 수집합니다:

import java.io.IOException
import kotlinx.coroutines.flow.retry

fun main() = runBlocking {
    var attempts = 0
    flow {
        attempts++
        if (attempts < 3) throw IOException("연결 끊김")
        emit("OK $attempts번 후")
    }
        .retry(3) { e -> e is IOException }
        .collect { println(it) }
}
OK 3번 후

Predicate { e is IOException }은 네트워크 오류로만 재시도할 수 있게 해줍니다. IllegalArgumentException 같은 비즈니스 오류는 재시도되지 않습니다. Predicate 내에서 delay()를 호출하여 백오프를 만들 수 있습니다.

7. StateFlow와 SharedFlow: hot stream

flow { }는 cold입니다. 하지만 때로는 collector와 독립적으로 존재하는 데이터 소스가 필요합니다. 예를 들어 장바구니 상태나 시스템 이벤트 같은 경우죠. 그럴 때는 hot stream이 필요합니다.

7.1. StateFlow

StateFlow는 항상 현재 값 하나를 유지합니다. 새로운 collector는 즉시 그 값을 받습니다:

import kotlinx.coroutines.cancel
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.launch

class CartService {
    private val _total = MutableStateFlow(0L)
    val total: StateFlow<Long> = _total.asStateFlow()

    fun add(price: Long) = _total.update { it + price }
}

fun main() = runBlocking {
    val cart = CartService()
    val job = launch {
        cart.total.collect { println("장바구니 총액: $it") }
    }
    delay(50)
    cart.add(100_000)
    delay(50)
    cart.add(250_000)
    delay(50)
    job.cancel()
}
장바구니 총액: 0
장바구니 총액: 100000
장바구니 총액: 350000

주의할 점:

  • _total 속성은 private이고 쓰기 가능합니다. 외부에서는 읽기 전용 StateFlow만 봅니다. 이는 상태를 캡슐화하는 표준 패턴입니다.
  • update { }는 원자적(atomic) 업데이트이므로 여러 스레드에서 동시에 add를 호출해도 안전합니다.
  • StateFlow는 절대 끝나지 않습니다. 따라서 job을 cancel해야 합니다. 그렇지 않으면 runBlocking이 계속 기다립니다.
  • StateFlow는 현재 값과 같은 값을 건너뜁니다. 또한 conflate입니다: 느린 collector는 최신 값만 보고 중간 단계는 보지 않습니다.

7.2. SharedFlow

SharedFlow는 이벤트에 적합합니다. 예를 들어 “주문이 결제되었습니다”처럼요. 여기서는 각 발행이 의미를 가지며 “현재 값”이라는 개념이 없습니다:

import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow

class PaymentService {
    private val _events = MutableSharedFlow<String>()
    val events: SharedFlow<String> = _events

    suspend fun pay(orderId: Int) {
        _events.emit("주문 #$orderId 결제 완료")
    }
}

fun main() = runBlocking {
    val payments = PaymentService()
    payments.pay(0)                       // 아직 청취자가 없음 -> 손실됨

    val job = launch { payments.events.collect { println("이메일: $it") } }
    delay(50)
    payments.pay(1)
    payments.pay(2)
    delay(50)
    job.cancel()
}
이메일: 주문 #1 결제 완료
이메일: 주문 #2 결제 완료

기본 구성(replay = 0)으로는 subscriber가 없을 때 발행된 이벤트가 손실됩니다. 늦게 도착한 collector가 최근 몇 가지 이벤트를 받으려면 MutableSharedFlow<String>(replay = 1)을 사용하세요.

또한 stateIn(scope, SharingStarted.WhileSubscribed(5_000), initialValue) 또는 shareIn(...)을 사용하여 cold Flow를 hot Flow로 변환할 수 있습니다. 이 경우 여러 collector가 하나의 upstream 실행을 공유합니다.

8. Flow 테스팅

라이브러리 kotlinx-coroutines-test는 runTest를 제공합니다. 이 함수는 가상 시간을 사용하여 delay를 건너뛰므로 테스트가 거의 즉시 실행됩니다:

import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.test.runTest
import org.assertj.core.api.Assertions.assertThat
import org.junit.jupiter.api.Test

class OrderFlowTest {

    @Test
    fun `100천 이상의 주문만 유지`() = runTest {
        val ids = orderStream()
            .filter { it.amount >= 100_000 }
            .map { it.id }
            .toList()

        assertThat(ids).containsExactly(2, 3)
    }

    @Test
    fun `결제 오류가 FAILED 상태로 변환됨`() = runTest {
        val states = paymentStatus()
            .catch { emit("FAILED") }
            .toList()

        assertThat(states).containsExactly("PENDING", "FAILED")
    }

    @Test
    fun `장바구니 값이 누적됨`() = runTest {
        val cart = CartService()
        cart.add(100_000)
        cart.add(250_000)

        assertThat(cart.total.value).isEqualTo(350_000L)
    }
}

유한한 Flow의 경우 toList()가 발행된 모든 값을 확인하는 가장 간단한 방법입니다. StateFlow의 경우 .value를 직접 읽습니다. StateFlow에 toList()를 호출하지 마세요. 이 Flow는 절대 끝나지 않으므로 테스트가 중단됩니다.

9. 흔한 실수

  • collect 이후의 코드가 실행되지 않습니다. collect는 suspend 함수이며 Flow가 끝날 때만 반환됩니다. StateFlow/SharedFlow의 경우 이는 절대 발생하지 않습니다. 각 무한 collect를 별도의 launch에 배치하세요.
  • 의도하지 않게 API를 여러 번 호출합니다. Cold Flow의 각 collector는 upstream을 다시 실행합니다. 공유가 필요하면 shareIn/stateIn을 사용하세요.
  • emit를 withContext로 감싸기. flowOn 또는 channelFlow로 대체하세요.
  • catch가 모든 오류를 포착한다고 생각하기. 이 연산자는 upstream 오류만 포착합니다. 처리 부분의 오류를 포착하려면 논리를 catch 앞에 위치한 onEach { }로 옮기고 매개변수 없이 collect()를 호출하세요.
  • 이벤트에 StateFlow 사용하기. 연속된 두 개의 동일한 이벤트가 하나로 병합됩니다. 이벤트에는 SharedFlow를 사용하세요.

참고: 하나의 결과만 있을 때는 Flow를 사용하면 안 됩니다. suspend fun findOrder(id: Int): Order 같은 간단한 함수가 하나의 요소만 발행하는 Flow<Order>보다 훨씬 읽기 쉽습니다.

10. 결론

이 글에서 Flow는 비동기 데이터 스트림이며 cold이고 lazy라는 것을 배웠습니다. map/filter로 데이터를 변환하고, flowOn으로 안전하게 스레드를 전환하고, catch, retry, onCompletion으로 오류를 처리하는 방법을 알게 되었습니다. 상태 및 이벤트 공유 필요성이 있을 때 StateFlow와 SharedFlow가 두 가지 주요 도구입니다. runTest는 빠르고 안정적으로 테스트하는 데 도움이 됩니다.

이제부터 combine, flatMapLatest, debounce, buffer 같은 고급 연산자를 시도하여 더 복잡한 데이터 파이프라인을 구축할 수 있습니다.

kotlincoroutinesflow비동기stateflow
저자의 앱HoanTien.proCashback on Shopee, TikTok Shop and Lazada (Vietnam)iOS · Android · Web

같은 주제의 글

Kotlin · 2026. 10. 26. · 9 분 읽기 · 44

Kotlin의 고차 함수와 람다

Kotlin에서 함수 타입 선언, 람다 전달 및 반환, 함수 참조, 클로저, 람다 with receiver 및 inline 사용법

Kotlin · 2026. 10. 25. · 8 분 읽기 · 59

Kotlin으로 Spring Boot 백엔드 만들기

Spring Boot 3를 Kotlin으로 설정하고, REST API와 JPA entity, 설정 그리고 테스트를 작성하며, Java에서 전환할 때 흔한 오류를 다룹니다.