kotlin-coroutines-flows

Kotlin Coroutines & Flows

適用於 Android 與 Kotlin Multiplatform (KMP) 專案的結構化併發、基於 Flow 的響應式串流（Reactive Streams）與 Coroutine 測試的最佳實踐模式。

何時啟用

• 使用 Kotlin Coroutines 編寫非同步程式碼時
• 使用 Flow、StateFlow 或 SharedFlow 處理響應式資料串流時
• 處理併發操作（如平行載入、防手震 Debounce、自動重試 Retry）時
• 針對 Coroutine 與 Flow 進行單元測試時
• 管理 Coroutine 作用域（Scope）與取消機制時

結構化併發 (Structured Concurrency)

作用域層級結構 (Scope Hierarchy)

Application
  └── viewModelScope (ViewModel)
        └── coroutineScope { } (結構化子作用域)
              ├── async { } (併發任務)
              └── async { } (併發任務)

務必使用結構化併發 — 絕不使用 GlobalScope：

// BAD — 缺乏作用域管控
GlobalScope.launch { fetchData() }

// GOOD — 綁定 ViewModel 生命週期
viewModelScope.launch { fetchData() }

// GOOD — 綁定 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 {
    // 當資料庫更新時自動重新發送 (emit)
    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 秒鐘 — 可在轉向等螢幕組態變更（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 運算子 (Flow Operators)

// 搜尋輸入防抖 (Debounce)
searchQuery
    .debounce(300)
    .distinctUntilChanged()
    .flatMapLatest { query -> repository.search(query) }
    .catch { emit(emptyList()) }
    .collect { results -> _state.update { it.copy(results = results) } }

// 指數退避重試 (Exponential Backoff Retry)
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("Item deleted"))
        }
    }
}

// 在 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) }

// I/O 密集型操作
withContext(Dispatchers.IO) { database.query() }

// 主執行緒 (UI) — viewModelScope 的預設值
withContext(Dispatchers.Main) { updateUi() }

在 KMP 專案中，請使用 Dispatchers.Default 與 Dispatchers.Main（所有平台皆支援）。Dispatchers.IO 僅限 JVM/Android 使用 — 在其他平台上請改用 Dispatchers.Default 或透過依賴注入（DI）提供。

取消機制 (Cancellation)

協作式取消 (Cooperative 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) }  // 無論正常結束或被取消，finally 區塊都會執行
    }
}

測試 (Testing)

使用 Turbine 測試 StateFlow

@Test
fun `search updates item list`() = 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 `parallel load completes correctly`() = runTest {
    val viewModel = DashboardViewModel(
        itemRepo = FakeItemRepo(),
        statsRepo = FakeStatsRepo()
    )

    viewModel.load()
    advanceUntilIdle()

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

模擬 Flow (Faking Flows)

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 })
    }
}

應避免的反模式 (Anti-Patterns)

• 使用 GlobalScope — 會導致 Coroutine 記憶體洩漏，且破壞結構化取消機制
• 在沒有 Scope 的 init {} 中收集 Flow — 請使用 viewModelScope.launch
• 將 MutableStateFlow 與可變集合（Mutable Collections）搭配使用 — 應始終傳入不可變的副本：_state.update { it.copy(list = it.list + newItem) }
• 捕捉 CancellationException — 應讓其繼續拋出以確保正常取消
• 使用 flowOn(Dispatchers.Main) 進行收集 — 收集時的 Dispatcher 應由呼叫端決定
• 在 @Composable 中建立 Flow 卻未使用 remember — 這會導致每次重組（Recomposition）時都重新建立 Flow

參考資料 (References)

參閱 Skill: compose-multiplatform-patterns 了解如何在 UI 層消費 Flow。
參閱 Skill: android-clean-architecture 了解 Coroutines 在架構分層中的定位。