Kotlin 協程與 Flow 是做什麼的?
在 Android 與 Kotlin Multiplatform 專案中,結構化併發、基於 Flow 的反應式串流以及協程測試的模式。
何時啟用
- 使用 Kotlin 協程撰寫非同步程式碼
- 使用 Flow、StateFlow 或 SharedFlow 處理反應式資料
- 處理並行操作(平行載入、防抖、重試)
- 測試協程與 Flow
- 管理協程範圍與取消
結構化併發
範圍階層
Application
└── viewModelScope (ViewModel)
└── coroutineScope { } (結構化子項)
├── async { } (並行任務)
└── async { } (並行任務)
永遠使用結構化併發 — 不要使用 GlobalScope:
// 不好
GlobalScope.launch { fetchData() }
// 好 — 範圍綁定 ViewModel 生命週期
viewModelScope.launch { fetchData() }
// 好 — 範圍綁定 Composable 生命週期
LaunchedEffect(key) { fetchData() }
平行分解
使用 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 模式
冷 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 秒 — 可在不重新啟動的情況下度過配置變更。
結合多個 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)
}
}
}
調度器
// 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 提供。
取消
合作式取消
長時間運行的迴圈必須檢查取消狀態:
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("查詢")
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
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— 讓它傳播以實現正確的取消 - 使用
flowOn(Dispatchers.Main)收集 — 收集的調度器是呼叫者的調度器 - 在
@Composable中不使用remember建立Flow— 每次重組都會重新建立 flow
參考資料
參閱技能:compose-multiplatform-patterns 以了解 Flow 的 UI 消費。
參閱技能:android-clean-architecture 以了解協程在各層中的定位。