Kotlin Coroutines & Flows は何をしますか？
Android および Kotlin Multiplatform プロジェクトにおける構造化された並行処理、Flow ベースのリアクティブストリーム、コルーチンテストのパターン。
有効な状況
- 非同期コードを Kotlin コルーチンで記述する場合
- Flow、StateFlow、または SharedFlow をリアクティブデータに使用する場合
- 並行処理（並列読み込み、デバウンス、リトライ）を扱う場合
- コルーチンと Flow をテストする場合
- コルーチンスコープとキャンセルを管理する場合
構造化された並行処理
スコープ階層
Application
└── viewModelScope (ViewModel)
└── coroutineScope { } (構造化された子)
├── async { } (並行タスク)
└── async { } (並行タスク)
常に構造化された並行処理を使用してください — GlobalScope は使用しないでください：
// 悪い例
GlobalScope.launch { fetchData() }
// 良い例 — ViewModel のライフサイクルにスコープ
viewModelScope.launch { fetchData() }
// 良い例 — コンポーザブルのライフサイクルにスコープ
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 パターン
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秒間アップストリームをアクティブに保ちます — 設定変更を再起動せずに生き延びます。
複数の 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("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)
}
}
}
ディスパッチャ
// 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 `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 のフェイク
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 を参照：コルーチンが各レイヤーでどのように使われるか。