Flow — その概要、Kotlinコルーチンにおけるコールドストリームとホットストリーム

KotlinのFlowとは？

Flowはkotlinx.coroutines.flowパッケージの型で、コールド非同期データストリームを表します。本質的に、Flowはemit()関数を介して値を出力し、成功または例外で終了するコルーチンシーケンスです。ストリームの収集は、サスペンド関数であるターミナル演算子collect()を介して実行されます。

コールドセマンティクス

コールドストリームは、flowビルダー内のコードがサブスクライバーごとに新たに実行されることを意味します。RxJavaのObservable.fromIterableも同様に動作します。新しいサブスクライバーは最初からすべての値を受け取ります。Flowでは、これはサスペンド関数collectを介して実装されており、データ収集の全期間中コルーチンをブロックします。

Flowビルダー

KotlinはFlowを作成するいくつかの方法を提供します: flow { } — emit()を使用した基本構造、flowOf(vararg values) — 固定値セット用、.asFlow() — コレクションとSequenceの拡張。すべてのビルダーはコールドです — データはターミナル演算子が呼び出されたときにのみ生成されます。

コールドストリームとホットストリーム

コールドストリームとホットストリームの区別は、リアクティブプログラミングの重要な概念です。コールドストリーム (Flow, Observable) は購読時にデータ生成を開始します。ホットストリーム (Channel, SharedFlow) は独立してデータを出力します — サブスクライバーは購読後に発生するもののみを受け取り、シーケンスの最初は受け取りません。

SharedFlowは複数のサブスクライバーを持てるホットFlowで、replayが設定されている場合、最近の値を再生できます。SharedFlowはイベント（一度きりの通知）に適しています。StateFlowは固定された状態値を持つそのバリアントで、新しいサブスクライバー用に最新の値をキャッシュします。

ChannelFlowは内部的にChannelを使用し、FlowとChannelのプロパティを組み合わせています。capacityを介したバッファリングとバックプレッシャーをサポートします。ChannelFlowは、コールバックAPIをリアクティブストリームに変換する際に便利で、値が異なるコルーチンから出力されます。

コールドとホットの変換

コールドFlowをホットSharedFlowに変換するには、shareIn(scope, started, replay)演算子を使用します。パラメータstartedは開始タイミングを制御します: SharingStarted.WhileSubscribed() — サブスクライバーがいる間アクティブ、Lazily — 最初のサブスクライバーで開始、Eagerly — 即時開始。逆変換 — ホットからコールド: StateFlow.asFlow()はコールドFlowを返し、collect時にStateFlowの現在の値を出力します。これはテストに便利です。

Flow演算子

Flowは、コルーチン内でサスペンド関数として動作する豊富な演算子セットを提供します。演算子はステートレスで、新しいFlowを返します — 元のストリームは変更されません。これにより、副作用なしで安全な変換チェーンを構築できます。

map演算子は、非同期または同期変換を介して各ストリーム値を変換します。filterは条件を満たす値のみを通過させます。catchはターミナル演算子の前に例外をキャッチし、ストリームの回復を可能にします。flatMapLatestは新しい値が到着すると前の出力をキャンセルします — RxのswitchMapと同様です。

Flowのdebounce演算子は、指定されたタイムアウトまで値の公開を遅延します。この間に新しい値が到着すると、タイマーがリセットされます。Androidでは、debounceは検索に使用されます: リクエストは300-400 msの一時停止後にのみ送信され、API呼び出しが3-5倍削減されます。

ターミナル演算子

collect()に加えて、Flowは他のターミナル演算子をサポートします: toList()はすべての値をリストに収集 — テストに便利、first()は最初の要素を返してストリームをキャンセル、single()は正確に1つの要素を期待します。fold(initial)は渡された関数を介して値を蓄積します。すべてのターミナル演算子はサスペンド関数であり、コルーチンまたは別のサスペンド関数内で呼び出す必要があります。

Flowコード例

最初の例 — map演算子による変換で数値を生成する基本Flow:

val numberFlow = flow {
    for (i in 1..5) {
        delay(500)
        emit(i)
    }
}

scope.launch {
    numberFlow
        .map { "数値: $it" }
        .collect { value ->
            println(value)
        }
}

2番目の例 — catchによるフィルタリングとエラー処理を含むストリーム変換:

flow {
    emit("data1")
    emit("data2")
    throw RuntimeException("network error")
}
    .catch { e ->
        emit("fallback_data")
    }
    .collect { value ->
        println(value)
    }

3番目の例 — Jetpack ComposeのリアクティブUIのためのViewModelでのStateFlow使用:

class SearchViewModel : ViewModel() {
    private val _query = MutableStateFlow("")
    val results: StateFlow<List<Result>> = _query
        .debounce(300)
        .flatMapLatest { query ->
            repository.search(query)
        }
        .catch { emit(emptyList()) }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), emptyList())

    fun onQueryChanged(query: String) {
        _query.value = query
    }
}

StateFlowとSharedFlow

StateFlowは単一の現在値を持つホットFlowです。最新の値をキャッシュし、新しいサブスクライバーに即座に渡します。StateFlowは状態のための監視可能なコンテナであり、equals比較をサポートします — 新しい値が現在の値と一致する場合、出力は発生しません。Jetpack ComposeはcollectAsState()を介してStateFlowを使用します。

SharedFlowは必須の初期値なしでより柔軟なホットFlowです。SharedFlowはreplay（新しいサブスクライバー用の値の数）、extraBufferCapacity（replayを超えるバッファ）、onBufferOverflow（オーバーフロー時の戦略）を介して設定されます。SharedFlowは一度きりのイベントに最適です: ナビゲーション、Snackbar、分析。

AndroidアーキテクチャにおけるFlowは、Googleによって主要なデータソース（層: Repository → UseCase → ViewModel）として推奨されています。LiveDataは柔軟性でFlowに劣ります: Flowはコルーチン、演算子、バックプレッシャーをサポートし、UI層の外部で動作します。LiveDataからFlowへの移行は、現代のAndroidプロジェクトでは標準的なプラクティスです。

ViewModelでFlowを使用する際は、正しいタイプを選択することが重要です。StateFlowは画面回転を生き残るべきUI状態に最適です。SharedFlowは再処理が許されないイベントに適しています — 例えばナビゲーション。lifecycleScopeでのcollect()を使用したFlowは、実行コンテキストの最大限の制御を提供しますが、画面離脱時に手動キャンセルが必要です。

Flowのテストはkotlinx-coroutines-testを介して行われます。このライブラリはTestDispatcherを提供します — 遅延（delay）を加速し、コルーチンの実行順序を制御できる仮想時間です。TestScope.runTest { }はFlowテストのための孤立した環境を作成します。toList()演算子は、ストリームが正しいデータシーケンスを出力したかを検証するために、タイムアウト付きですべてのflow値を収集するテストでよく使用されます。

FlowはRoom（Android DBライブラリ）とよく統合されます: DAOメソッドはFlow<List<Entity>>を返せます。Roomはテーブル変更時に自動的に新しい値を出力します — 手動トリガーなしでUIが更新されます。これは、内部的にcallbackFlowを使用するInvalidationTrackerを介して実装されます。このアプローチはLiveDataの必要性を排除し、データ層を完全にコルーチン指向にします。Jetpack ComposeはcollectAsState()を介してStateFlowにサブスクライブし、データが変更されたコンポーネントのみを再描画します — これはLiveData指向のアーキテクチャでは達成できないパフォーマンスを提供します。DataStore（SharedPreferencesの代替）もFlow<Preferences>を返し、手動更新トリガーなしでアプリケーション設定のリアクティブな読み取りを提供します。

Flowは追加ライブラリなしでJVM上のkotlinx-coroutines-coreを介してプロセス間通信をサポートします。例えば、Ktorのサーバーアプリケーションでは、Flowは着信WebSocketメッセージストリームを表せます。各メッセージはストリームに出力され、演算子を介したフィルタリングと集約を経て、結果がクライアントに送信されます。このアプローチは、KotlinプロジェクトにおけるReactorやRxJavaのようなリアクティブライブラリを置き換えます。

既存のRxJavaコードとのFlow互換性はkotlinx-coroutines-rx3モジュールによって提供されます。拡張関数Flow.asObservable()はFlowをRxJava 3のObservableに変換します。逆変換 — CompletableSource.asFlow()、Observable.asFlow()。これにより、RxJavaからコルーチンへの移行が簡素化されます: 一部の層をRxJavaに残したまま、段階的にプロジェクトを書き換えられます。変換時には、コールド/ホットセマンティクスの違いを考慮する必要があります: Observableはコールドにもホットにもなれますが、Flowは通常のFlowでは常にコールド、SharedFlowではホットです。

Flowのエラー処理とテスト

Flowのエラー処理には特徴があります: ターミナル演算子の前にflowビルダー内で例外が発生した場合、それはcatchに伝播されます。ビルダー後の演算子で例外が発生した場合、その演算子の後のcatchが捕捉します。retryWhenは条件付きでサブスクリプションを再試行できます: ネットワークエラー時は最大3回再試行、しかしCancellationExceptionでは再試行しません。Flowは状態を保存しないため、状態依存のエラーを排除します — これはSubjectが内部状態を保存するObservableと比較してデバッグを簡素化します。

kotlinx-coroutines-testを使用したFlowテストは、遅延をシミュレートするためにTestDispatcherを使用します。TurbineはFlowテストのための人気のあるコミュニティライブラリです: test { }はFlowを起動し、awaitItem()は次の値を待機し、awaitComplete()は完了を待機します。Turbineはデフォルトのタイムアウトを追加し、テストのハングアップを防止します。StateFlowのテストには、時系列順の値検証とともに.testIn(scope)を使用します。