kotlin-coroutines-flows

Kotlin 코루틴 및 Flow

Android 및 Kotlin 멀티플랫폼(KMP) 프로젝트에서의 구조화된 동시성, Flow 기반 리액티브 스트림 및 코루틴 테스트를 위한 패턴입니다.

활성화 시점

• Kotlin 코루틴으로 비동기 코드를 작성할 때
• 리액티브 데이터를 위해 Flow, StateFlow 또는 SharedFlow를 사용할 때
• 동시 작업(병렬 로딩, 디바운스, 재시도)을 처리할 때
• 코루틴 및 Flow를 테스트할 때
• 코루틴 스코프 및 취소를 관리할 때

구조화된 동시성 (Structured Concurrency)

스코프 계층 구조

Application
  └── viewModelScope (ViewModel)
        └── coroutineScope { } (구조화된 자식)
              ├── async { } (동시 작업)
              └── async { } (동시 작업)

항상 구조화된 동시성을 사용하십시오 — GlobalScope는 절대 사용하지 마십시오:

// 나쁨
GlobalScope.launch { fetchData() }

// 좋음 — ViewModel 생명주기에 종속됨
viewModelScope.launch { fetchData() }

// 좋음 — Composable 생명주기에 종속됨
LaunchedEffect(key) { fetchData() }

병렬 분해 (Parallel Decomposition)

병렬 작업을 위해 coroutineScope + async를 사용하십시오:

suspend fun loadDashboard(): Dashboard = coroutineScope {
    val items = async { itemRepository.getRecent() }
    val stats = async { statsRepository.getToday() }
    val profile = async { userRepository.getCurrent() }
    Dashboard(
        items = items.await(),
        stats = stats.await(),
        profile = profile.await()
    )
}

SupervisorScope

자식의 실패가 형제 코루틴을 취소하지 않아야 할 때 supervisorScope를 사용하십시오:

suspend fun syncAll() = supervisorScope {
    launch { syncItems() }       // 여기서 실패해도 syncStats는 취소되지 않음
    launch { syncStats() }
    launch { syncSettings() }
}

Flow 패턴

Cold Flow — 단발성 이벤트를 스트림으로 변환

fun observeItems(): Flow<List<Item>> = flow {
    // 데이터베이스가 변경될 때마다 다시 방출
    itemDao.observeAll()
        .map { entities -> entities.map { it.toDomain() } }
        .collect { emit(it) }
}

UI 상태를 위한 StateFlow

class DashboardViewModel(
    observeProgress: ObserveUserProgressUseCase
) : ViewModel() {
    val progress: StateFlow<UserProgress> = observeProgress()
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5_000),
            initialValue = UserProgress.EMPTY
        )
}

WhileSubscribed(5_000)는 마지막 구독자가 떠난 후에도 5초 동안 상류(upstream)를 활성 상태로 유지하여, 재시작 없이 구성 변경(configuration changes)에서 살아남을 수 있게 합니다.

여러 Flow 결합

val uiState: StateFlow<HomeState> = combine(
    itemRepository.observeItems(),
    settingsRepository.observeTheme(),
    userRepository.observeProfile()
) { items, theme, profile ->
    HomeState(items = items, theme = theme, profile = profile)
}.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), HomeState())

Flow 연산자

// 검색 입력 디바운스
searchQuery
    .debounce(300)
    .distinctUntilChanged()
    .flatMapLatest { query -> repository.search(query) }
    .catch { emit(emptyList()) }
    .collect { results -> _state.update { it.copy(results = results) } }

// 지수 백오프를 사용한 재시도
fun fetchWithRetry(): Flow<Data> = flow { emit(api.fetch()) }
    .retryWhen { cause, attempt ->
        if (cause is IOException && attempt < 3) {
            delay(1000L * (1 shl attempt.toInt()))
            true
        } else {
            false
        }
    }

단발성 이벤트를 위한 SharedFlow

class ItemListViewModel : ViewModel() {
    private val _effects = MutableSharedFlow<Effect>()
    val effects: SharedFlow<Effect> = _effects.asSharedFlow()

    sealed interface Effect {
        data class ShowSnackbar(val message: String) : Effect
        data class NavigateTo(val route: String) : Effect
    }

    private fun deleteItem(id: String) {
        viewModelScope.launch {
            repository.delete(id)
            _effects.emit(Effect.ShowSnackbar("아이템 삭제됨"))
        }
    }
}

// Composable에서 수집
LaunchedEffect(Unit) {
    viewModel.effects.collect { effect ->
        when (effect) {
            is Effect.ShowSnackbar -> snackbarHostState.showSnackbar(effect.message)
            is Effect.NavigateTo -> navController.navigate(effect.route)
        }
    }
}

디스패처 (Dispatchers)

// CPU 집약적인 작업
withContext(Dispatchers.Default) { parseJson(largePayload) }

// IO 관련 작업
withContext(Dispatchers.IO) { database.query() }

// 메인 스레드 (UI) — viewModelScope의 기본값
withContext(Dispatchers.Main) { updateUi() }

KMP에서는 모든 플랫폼에서 사용 가능한 Dispatchers.Default 및 Dispatchers.Main을 사용하십시오. Dispatchers.IO는 JVM/Android 전용입니다 — 다른 플랫폼에서는 Dispatchers.Default를 사용하거나 DI를 통해 제공하십시오.

취소 (Cancellation)

협력적 취소

장시간 실행되는 루프는 반드시 취소 여부를 확인해야 합니다:

suspend fun processItems(items: List<Item>) = coroutineScope {
    for (item in items) {
        ensureActive()  // 취소된 경우 CancellationException 발생
        process(item)
    }
}

try/finally를 사용한 정리

viewModelScope.launch {
    try {
        _state.update { it.copy(isLoading = true) }
        val data = repository.fetch()
        _state.update { it.copy(data = data) }
    } finally {
        _state.update { it.copy(isLoading = false) }  // 취소되더라도 항상 실행됨
    }
}

테스트

Turbine을 사용한 StateFlow 테스트

@Test
fun `검색 시 아이템 목록이 업데이트됨`() = runTest {
    val fakeRepository = FakeItemRepository().apply { emit(testItems) }
    val viewModel = ItemListViewModel(GetItemsUseCase(fakeRepository))

    viewModel.state.test {
        assertEquals(ItemListState(), awaitItem())  // 초기 상태

        viewModel.onSearch("query")
        val loading = awaitItem()
        assertTrue(loading.isLoading)

        val loaded = awaitItem()
        assertFalse(loaded.isLoading)
        assertEquals(1, loaded.items.size)
    }
}

TestDispatcher를 사용한 테스트

@Test
fun `병렬 로드가 올바르게 완료됨`() = runTest {
    val viewModel = DashboardViewModel(
        itemRepo = FakeItemRepo(),
        statsRepo = FakeStatsRepo()
    )

    viewModel.load()
    advanceUntilIdle()

    val state = viewModel.state.value
    assertNotNull(state.items)
    assertNotNull(state.stats)
}

Flow 페이킹 (Faking)

class FakeItemRepository : ItemRepository {
    private val _items = MutableStateFlow<List<Item>>(emptyList())

    override fun observeItems(): Flow<List<Item>> = _items

    fun emit(items: List<Item>) { _items.value = items }

    override suspend fun getItemsByCategory(category: String): Result<List<Item>> {
        return Result.success(_items.value.filter { it.category == category })
    }
}

피해야 할 안티 패턴

• GlobalScope 사용 — 코루틴 누수 발생 및 구조화된 취소 불가
• 스코프 없이 init {}에서 Flow 수집 — viewModelScope.launch 사용 권장
• 가변 컬렉션과 함께 MutableStateFlow 사용 — 항상 불변 복사본 사용: _state.update { it.copy(list = it.list + newItem) }
• CancellationException 포착 — 적절한 취소를 위해 전파되도록 둠
• 수집(collect)을 위해 flowOn(Dispatchers.Main) 사용 — 수집 디스패처는 호출자의 디스패처임
• remember 없이 @Composable에서 Flow 생성 — 리컴포지션마다 Flow를 다시 생성함

참조

Flow의 UI 사용에 대해서는 compose-multiplatform-patterns 스킬을 참조하십시오. 코루틴이 계층 구조에서 어디에 위치하는지에 대해서는 android-clean-architecture 스킬을 참조하십시오.