Kotlin Flowの使い方 StateFlowとSharedFlowの違い

Kotlin の Flow とは、時間をおいて順番に届く複数の値を、コルーチンの上で扱うための型です。 flow { } で作る Flow は「collect されるたびに最初から流れるコールドフロー」、StateFlow と SharedFlow は「購読者がいなくても値が流れ続けるホットフロー」で、StateFlow は最新の値を1つ持つ状態の入れ物、SharedFlow は起きた出来事を配る放送と考えると選びやすくなります。

この記事のコードはすべて runBlocking だけで動くコンソールプログラムです。画面のフレームワークを使わずに、3種類の違いをタイムスタンプ付きの出力で比べ、演算子の一覧、stateIn / shareIn の started の違い、collect の次の行に進まない現象と launchIn での書き分けまで、実際の出力を添えて説明します。

確認環境：Ubuntu 24.04 / Kotlin 2.4.20（kotlinc）/ OpenJDK 21.0.10 / kotlinx-coroutines-core 1.8.0。掲載した出力はすべてこの環境で実行したものです（ミリ秒の値は実行のたびに少し変わります）。

Flow・StateFlow・SharedFlow の早見表

迷ったときの選び方：「今の値」が意味を持つなら StateFlow、「起きたこと」を1回ずつ伝えたいなら SharedFlow、呼ばれるたびに処理をやり直したいなら flow { } です。

準備：依存の追加と実行コマンド

Flow は標準ライブラリではなく kotlinx.coroutines に入っています。Gradle（Kotlin DSL）では、公式リポジトリの README にある次の1行を追加します（README に記載の最新版は 1.11.0 です）。

dependencies {
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.11.0")
}

この記事では Gradle を使わず、kotlinc に jar を直接渡して実行しました。-Dstdout.encoding=UTF-8 は日本語の出力を UTF-8 にそろえるための指定です。

kotlinc Basic.kt -cp kotlinx-coroutines-core-jvm.jar -include-runtime -d basic.jar
java -Dstdout.encoding=UTF-8 -cp basic.jar:kotlinx-coroutines-core-jvm.jar BasicKt

flow ビルダーと collect：まず動かす

flow { } の中で emit(値) を呼ぶと値が1つ流れ、受け取る側は collect { } で1つずつ処理します。map や filter などの演算子は、collect の前にいくつでもつなげられます。

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.*

fun numbers(): Flow<Int> = flow {
    println("  flow 開始")
    for (i in 1..5) {
        delay(100)
        emit(i)
    }
}

fun main() = runBlocking {
    println("1) collect でそのまま受け取る")
    numbers().collect { println("  受信 $it") }

    println("2) 演算子をつなぐ")
    numbers()
        .filter { it % 2 == 1 }
        .map { it * 10 }
        .onEach { println("  onEach $it") }
        .take(2)
        .collect { println("  受信 $it") }

    println("3) 終端演算子で値にする")
    println("  toList = " + numbers().toList())
    println("  first  = " + numbers().first())
    println("  fold   = " + numbers().fold(0) { acc, v -> acc + v })

    println("4) ほかの作り方")
    println("  flowOf = " + flowOf("a", "b").toList())
    println("  asFlow = " + listOf(1, 2, 3).asFlow().map { it * it }.toList())
}

1) collect でそのまま受け取る
  flow 開始
  受信 1
  受信 2
  受信 3
  受信 4
  受信 5
2) 演算子をつなぐ
  flow 開始
  onEach 10
  受信 10
  onEach 30
  受信 30
3) 終端演算子で値にする
  flow 開始
  toList = [1, 2, 3, 4, 5]
  flow 開始
  first  = 1
  flow 開始
  fold   = 15
4) ほかの作り方
  flowOf = [a, b]
  asFlow = [1, 4, 9]

読み取れることは3つあります。

• flow 開始 が collect・toList・first のたびに表示されている：Flow は呼ばれるたびに最初から実行されます。
• take(2) は2個受け取った時点で上流を止める：3個目以降の emit は実行されません。
• collect・toList・first・fold は suspend 関数：これらを終端演算子と呼び、呼んだ時点で初めて Flow が動きます。map や filter は Flow を返すだけで、それだけでは何も実行されません。

コールドフロー：collect のたびに最初から流れる

Flow を作ってから 300 ミリ秒待ち、A と B が順番に collect します。行頭の数字はプログラム開始からの経過ミリ秒です。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

val t0 = System.currentTimeMillis()
fun log(msg: String) = println("[%4d ms] %s".format(System.currentTimeMillis() - t0, msg))

fun main() = runBlocking {
    val cold = flow {
        log("flow ブロック開始")
        for (i in 1..3) {
            delay(100)
            emit(i)
        }
    }
    log("flow を作っただけ（まだ何も流れない）")
    delay(300)

    log("A が collect")
    cold.collect { log("A 受信 $it") }

    log("B が collect")
    cold.collect { log("B 受信 $it") }
}

[ 104 ms] flow を作っただけ（まだ何も流れない）
[ 439 ms] A が collect
[ 447 ms] flow ブロック開始
[ 591 ms] A 受信 1
[ 691 ms] A 受信 2
[ 792 ms] A 受信 3
[ 793 ms] B が collect
[ 795 ms] flow ブロック開始
[ 895 ms] B 受信 1
[ 996 ms] B 受信 2
[1096 ms] B 受信 3

作っただけの 300 ミリ秒間は何も起きず、A が collect した 447 ms に初めてブロックが動いています。 B が collect すると、もう一度 flow ブロック開始 が出て 1 から流れ直します。A が受け取り終わったあとに来た B も、1・2・3 をすべて受け取れます。これがコールドフローです。公式ドキュメントも、コールドフローのブロックは collect されるまで動かず、新しい collect ごとに新しい実行が始まると説明しています。

ホットフロー：StateFlow と SharedFlow は購読前の値を replay 分だけ受け取る

同じ流れを SharedFlow と StateFlow で試します。早い購読者は最初から、遅い購読者は 1・2・3 が流れたあとに collect を始めます。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

val t0 = System.currentTimeMillis()
fun log(msg: String) = println("[%4d ms] %s".format(System.currentTimeMillis() - t0, msg))

suspend fun demo(name: String, shared: MutableSharedFlow<Int>) = coroutineScope {
    log("--- $name ---")
    val early = launch { shared.collect { log("  早い購読者 受信 $it") } }
    delay(50)
    for (i in 1..3) {
        log("emit $i")
        shared.emit(i)
        delay(100)
    }
    val late = launch { shared.collect { log("  遅い購読者 受信 $it") } }
    delay(50)
    log("emit 4")
    shared.emit(4)
    delay(100)
    early.cancel()
    late.cancel()
}

fun main() = runBlocking {
    demo("SharedFlow replay=0", MutableSharedFlow(replay = 0))
    demo("SharedFlow replay=2", MutableSharedFlow(replay = 2))

    log("--- StateFlow 初期値 0 ---")
    val state = MutableStateFlow(0)
    val early = launch { state.collect { log("  早い購読者 受信 $it") } }
    delay(50)
    for (i in 1..3) {
        log("value = $i")
        state.value = i
        delay(100)
    }
    val late = launch { state.collect { log("  遅い購読者 受信 $it") } }
    delay(50)
    log("value = 4")
    state.value = 4
    delay(100)
    log("今の値 state.value = ${state.value}")
    early.cancel()
    late.cancel()
}

[ 156 ms] --- SharedFlow replay=0 ---
[ 266 ms] emit 1
[ 268 ms]   早い購読者 受信 1
[ 369 ms] emit 2
[ 369 ms]   早い購読者 受信 2
[ 469 ms] emit 3
[ 470 ms]   早い購読者 受信 3
[ 627 ms] emit 4
[ 628 ms]   早い購読者 受信 4
[ 628 ms]   遅い購読者 受信 4
[ 732 ms] --- SharedFlow replay=2 ---
[ 782 ms] emit 1
[ 783 ms]   早い購読者 受信 1
[ 887 ms] emit 2
[ 887 ms]   早い購読者 受信 2
[ 987 ms] emit 3
[ 987 ms]   早い購読者 受信 3
[1088 ms]   遅い購読者 受信 2
[1088 ms]   遅い購読者 受信 3
[1138 ms] emit 4
[1139 ms]   早い購読者 受信 4
[1139 ms]   遅い購読者 受信 4
[1240 ms] --- StateFlow 初期値 0 ---
[1243 ms]   早い購読者 受信 0
[1292 ms] value = 1
[1292 ms]   早い購読者 受信 1
[1393 ms] value = 2
[1393 ms]   早い購読者 受信 2
[1494 ms] value = 3
[1494 ms]   早い購読者 受信 3
[1595 ms]   遅い購読者 受信 3
[1645 ms] value = 4
[1646 ms]   早い購読者 受信 4
[1646 ms]   遅い購読者 受信 4
[1746 ms] 今の値 state.value = 4

遅い購読者が受け取った値だけを抜き出すと、違いがはっきりします。

ホットフローでは、値は購読者の有無に関係なく流れ、あとから来た購読者に届くのは replay の分だけです。 StateFlow は初期値を持つので、早い購読者も collect した直後（1243 ms）に 0 を受け取っています。

StateFlow は同じ値を間引き、SharedFlow はそのまま流す

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val state = MutableStateFlow("A")
    val shared = MutableSharedFlow<String>()

    val j1 = launch { state.collect { println("StateFlow  受信 $it") } }
    val j2 = launch { shared.collect { println("SharedFlow 受信 $it") } }
    delay(50)

    for (v in listOf("A", "B", "B", "C")) {
        println("送る $v")
        state.value = v
        shared.emit(v)
        delay(50)
    }

    println("--- 待たずに続けて代入 ---")
    state.value = "X"
    state.value = "Y"
    state.value = "Z"
    delay(50)

    j1.cancel()
    j2.cancel()
}

StateFlow  受信 A
送る A
SharedFlow 受信 A
送る B
StateFlow  受信 B
SharedFlow 受信 B
送る B
SharedFlow 受信 B
送る C
StateFlow  受信 C
SharedFlow 受信 C
--- 待たずに続けて代入 ---
StateFlow  受信 Z

• 1行目の StateFlow 受信 A は初期値です。そのあと A を代入しても、今の値と同じなので StateFlow の購読者には届きません。2回目の B も同様です。SharedFlow は同じ値でも毎回届けています。
• StateFlow に X・Y・Z を続けて代入すると、届いたのは最後の Z だけです。StateFlow は「最新の値」を伝えるもので、途中の値をすべて届ける用途には向きません。1件ずつ確実に伝えたいイベントには SharedFlow を使います。

公式の API ドキュメントは、StateFlow の値は Any.equals の比較でまとめられ distinctUntilChanged と似た動きをすること、StateFlow は「replay = 1、あふれたら最も前の値を捨てる設定の SharedFlow に初期値を入れ、distinctUntilChanged を付けたもの」と同じ動きをすることを説明しています。data class を値にする場合は、中身が同じなら流れないので、copy で内容を変えた新しいインスタンスを代入します。

演算子の一覧表

flowOf(3, 1, 1, 2, 2, 5) を入力にして実行した結果です。表の右列は実際の出力をそのまま載せています。

途中でつなぐ演算子（Flow を返す）

終端演算子（suspend 関数。呼ぶと Flow が動く）

時間に関わる演算子

30 ミリ秒おきに 1〜5 を流す Flow を、処理に時間がかかる受け取り側で試した結果です。

debounce と sample は 1.8.0 では @OptIn(FlowPreview::class)、flatMapConcat は @OptIn(ExperimentalCoroutinesApi::class) を付けて使いました。上の3つの表を作ったコードの全文です。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun src() = flowOf(3, 1, 1, 2, 2, 5)

@OptIn(ExperimentalCoroutinesApi::class, FlowPreview::class)
fun main() = runBlocking {
    println("map                  " + src().map { it * 2 }.toList())
    println("filter               " + src().filter { it > 1 }.toList())
    println("mapNotNull           " + src().mapNotNull { if (it > 2) "v$it" else null }.toList())
    println("transform            " + src().transform { if (it > 2) { emit(it); emit(-it) } }.toList())
    println("take(2)              " + src().take(2).toList())
    println("takeWhile            " + src().takeWhile { it != 2 }.toList())
    println("drop(2)              " + src().drop(2).toList())
    println("dropWhile            " + src().dropWhile { it != 2 }.toList())
    println("distinctUntilChanged " + src().distinctUntilChanged().toList())
    println("withIndex            " + src().take(2).withIndex().toList())
    println("scan                 " + src().scan(0) { a, v -> a + v }.toList())
    println("runningReduce        " + src().runningReduce { a, v -> a + v }.toList())
    println("onStart              " + src().take(1).onStart { emit(0) }.toList())
    println("onCompletion         " + src().take(1).onCompletion { emit(99) }.toList())
    println("onEmpty              " + emptyFlow<Int>().onEmpty { emit(-1) }.toList())
    println("zip                  " + src().zip(flowOf("a", "b")) { n, s -> "$n$s" }.toList())
    println("combine              " + flowOf(1, 2).combine(flowOf("a")) { n, s -> "$n$s" }.toList())
    println("merge                " + merge(flowOf(1, 2), flowOf(10)).toList())
    println("flatMapConcat        " + flowOf(1, 2).flatMapConcat { flowOf(it, it * 10) }.toList())
    println("--- 終端演算子 ---")
    println("toList               " + src().toList())
    println("toSet                " + src().toSet())
    println("first                " + src().first())
    println("first { }            " + src().first { it < 3 })
    println("firstOrNull          " + emptyFlow<Int>().firstOrNull())
    println("last                 " + src().last())
    println("lastOrNull           " + emptyFlow<Int>().lastOrNull())
    println("single               " + flowOf(7).single())
    println("singleOrNull         " + src().singleOrNull())
    println("count                " + src().count())
    println("count { }            " + src().count { it > 1 })
    println("reduce               " + src().reduce { a, v -> a + v })
    println("fold                 " + src().fold(100) { a, v -> a + v })
    println("--- 空の Flow で例外になるもの ---")
    for ((name, f) in listOf<Pair<String, suspend () -> Any?>>(
        "first" to { emptyFlow<Int>().first() },
        "single(複数)" to { src().single() },
        "reduce" to { emptyFlow<Int>().reduce { a, v -> a + v } },
    )) {
        try { f() } catch (e: Exception) { println("$name -> ${e::class.simpleName}: ${e.message}") }
    }
    println("--- 時間に関わる演算子 ---")
    val slow = flow { for (i in 1..5) { emit(i); delay(30) } }
    val got = mutableListOf<Int>()
    slow.collectLatest { delay(50); got += it }
    println("collectLatest        $got")
    println("conflate             " + slow.conflate().onEach { delay(100) }.toList())
    println("buffer               " + slow.buffer().onEach { delay(100) }.toList())
    println("debounce(50)         " + flow { emit(1); delay(10); emit(2); delay(100); emit(3) }.debounce(50).toList())
    println("sample(50)           " + slow.sample(50).toList())
}

flowOn：上流を動かすスレッドを切り替える

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun th() = Thread.currentThread().name

fun main() = runBlocking {
    flow {
        println("emit   : ${th()}")
        emit(1)
    }
        .map { println("map 上 : ${th()}"); it * 2 }
        .flowOn(Dispatchers.Default)
        .map { println("map 下 : ${th()}"); it + 1 }
        .collect { println("collect: ${th()} 値=$it") }
}

emit   : DefaultDispatcher-worker-1
map 上 : DefaultDispatcher-worker-1
map 下 : main
collect: main 値=3

flowOn が切り替えるのは、自分より上に書いた部分（上流）だけです。 flowOn より下の map と collect は、collect を呼んだ側のスレッド（ここでは main）のままです。重い読み込みを別スレッドに移し、結果の表示は呼び出し側で行う、という分担が1行で書けます。

flow { } の中で withContext を使って emit すると、次の例外になります。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    flow {
        withContext(Dispatchers.Default) {
            emit(1)
        }
    }.collect { println(it) }
}

Exception in thread "main" java.lang.IllegalStateException: Flow invariant is violated:
        Flow was collected in [BlockingCoroutine{Active}@30bad64, BlockingEventLoop@41fda58c],
        but emission happened in [DispatchedCoroutine{Active}@47cdf8b, Dispatchers.Default].
        Please refer to 'flow' documentation or use 'flowOn' instead
    at kotlinx.coroutines.flow.internal.SafeCollector_commonKt.checkContext(SafeCollector.common.kt:84)
…（省略）…

メッセージの最後にあるとおり、スレッドを変えたいときは withContext ではなく flowOn を使います。公式ドキュメントでも、flow() ビルダーは動いているコンテキストと同じコンテキストから emit する必要があり、別のコンテキストで動かしたいときは flowOn か channelFlow() を使うと案内されています。

catch・onCompletion・retry：例外の扱い方

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun source(): Flow<Int> = flow {
    emit(1)
    emit(2)
    throw IllegalStateException("読み込みが止まりました")
}

fun main() = runBlocking {
    println("1) catch で受けて代わりの値を流す")
    source()
        .catch { e -> println("  catch: ${e.message}"); emit(-1) }
        .onCompletion { cause -> println("  onCompletion cause=$cause") }
        .collect { println("  受信 $it") }

    println("2) catch より下流の例外は catch に届かない")
    try {
        source()
            .catch { e -> println("  catch: ${e.message}") }
            .collect { v ->
                check(v < 2) { "collect の中で検査に引っかかった: $v" }
                println("  受信 $v")
            }
    } catch (e: IllegalStateException) {
        println("  try-catch: ${e.message}")
    }

    println("3) onEach に処理を移すと catch で受けられる")
    source()
        .onEach { v ->
            check(v < 2) { "onEach の中で検査に引っかかった: $v" }
            println("  受信 $v")
        }
        .catch { e -> println("  catch: ${e.message}") }
        .collect()

    println("4) retry でやり直す")
    var attempt = 0
    flow {
        attempt++
        println("  $attempt 回目")
        if (attempt < 3) throw IllegalStateException("まだ準備中")
        emit("OK")
    }
        .retry(2)
        .collect { println("  受信 $it") }
}

1) catch で受けて代わりの値を流す
  受信 1
  受信 2
  catch: 読み込みが止まりました
  受信 -1
  onCompletion cause=null
2) catch より下流の例外は catch に届かない
  受信 1
  try-catch: collect の中で検査に引っかかった: 2
3) onEach に処理を移すと catch で受けられる
  受信 1
  catch: onEach の中で検査に引っかかった: 2
4) retry でやり直す
  1 回目
  2 回目
  3 回目
  受信 OK

catch が受けるのは、自分より上流で起きた例外だけです。 2) のように collect のラムダで起きた例外は catch を通りません。公式ドキュメントも、collect に渡したラムダの例外は try-catch で collect を囲んで扱うと説明しています。

collect の次の行が実行されないとき：launchIn での書き分け

StateFlow と SharedFlow は完了しないので、collect は cancel されるまで戻ってきません。collect の下に書いた行は実行されないままになります。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val state = MutableStateFlow(0)

    val job = launch {
        println("collect を始める")
        state.collect { println("  受信 $it") }
        println("この行は実行されない")
    }

    delay(100)
    state.value = 1
    delay(100)
    println("job.isActive = ${job.isActive}  isCompleted = ${job.isCompleted}")
    job.cancel()
    job.join()
    println("cancel 後 isCompleted = ${job.isCompleted}")
}

collect を始める
  受信 0
  受信 1
job.isActive = true  isCompleted = false
cancel 後 isCompleted = true

この行は実行されない は表示されず、collect を呼んだコルーチンは cancel するまで動いたままです。公式の API ドキュメントにも、StateFlow は完了せず、collect の呼び出しは正常には終わらないと書かれています。

この性質から、1つの launch に collect を2つ並べると、2つ目が始まりません。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val count = MutableStateFlow(0)
    val message = MutableSharedFlow<String>()

    println("--- 1つの launch に collect を2つ並べる ---")
    val bad = launch {
        count.collect { println("  count 受信 $it") }
        message.collect { println("  message 受信 $it") }   // ここまで進まない
    }
    delay(50)
    count.value = 1
    message.emit("こんにちは")
    delay(50)
    bad.cancel()

    println("--- launchIn で1本ずつ起動する ---")
    val scope = CoroutineScope(coroutineContext + Job())
    count.onEach { println("  count 受信 $it") }.launchIn(scope)
    message.onEach { println("  message 受信 $it") }.launchIn(scope)
    println("launchIn の次の行はすぐ実行される")
    delay(50)
    count.value = 2
    message.emit("こんにちは")
    delay(50)
    scope.cancel()
}

--- 1つの launch に collect を2つ並べる ---
  count 受信 0
  count 受信 1
--- launchIn で1本ずつ起動する ---
launchIn の次の行はすぐ実行される
  count 受信 1
  count 受信 2
  message 受信 こんにちは

前半では message 受信 が一度も出ていません。後半のように onEach { 処理 }.launchIn(scope) と書くと、collect が別のコルーチンで始まり、次の行へすぐ進みます。launchIn(scope) は scope.launch { flow.collect() } を短く書いたもので、戻り値の Job や scope の cancel で止められます。

stateIn と shareIn：コールドフローをホットに変える

shareIn は Flow を SharedFlow に、stateIn は StateFlow に変えます。どちらも、上流を動かす CoroutineScope と、いつ動かし始めるかを決める started を渡します。started の3種類を、同じ手順（250 ミリ秒後に購読、さらに 250 ミリ秒後に購読をやめる）で比べます。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

val t0 = System.currentTimeMillis()
fun log(msg: String) = println("[%4d ms] %s".format(System.currentTimeMillis() - t0, msg))

fun ticker(name: String): Flow<Int> = flow {
    log("$name: 上流 開始")
    try {
        var i = 0
        while (true) {
            i++
            log("$name: 上流 emit $i")
            emit(i)
            delay(100)
        }
    } finally {
        log("$name: 上流 停止")
    }
}

suspend fun run(name: String, started: SharingStarted) = coroutineScope {
    val scope = CoroutineScope(coroutineContext + Job())
    val t = System.currentTimeMillis()
    log("=== $name ===")
    val shared = ticker(name).shareIn(scope, started, replay = 1)
    delay(250)
    log("$name: 購読者が collect 開始")
    val job = launch { shared.collect { log("$name:   購読者 受信 $it") } }
    delay(250)
    log("$name: 購読者が cancel")
    job.cancel()
    delay(250)
    log("$name: scope を cancel")
    scope.cancel()
    delay(50)
}

fun main() = runBlocking {
    run("Eagerly", SharingStarted.Eagerly)
    run("Lazily", SharingStarted.Lazily)
    run("WhileSubscribed", SharingStarted.WhileSubscribed())
}

[ 129 ms] === Eagerly ===
[ 191 ms] Eagerly: 上流 開始
[ 192 ms] Eagerly: 上流 emit 1
[ 300 ms] Eagerly: 上流 emit 2
[ 402 ms] Eagerly: 上流 emit 3
[ 433 ms] Eagerly: 購読者が collect 開始
[ 478 ms] Eagerly:   購読者 受信 3
[ 503 ms] Eagerly: 上流 emit 4
[ 503 ms] Eagerly:   購読者 受信 4
[ 604 ms] Eagerly: 上流 emit 5
[ 605 ms] Eagerly:   購読者 受信 5
[ 704 ms] Eagerly: 上流 emit 6
[ 705 ms] Eagerly:   購読者 受信 6
[ 709 ms] Eagerly: 購読者が cancel
[ 805 ms] Eagerly: 上流 emit 7
[ 908 ms] Eagerly: 上流 emit 8
[ 963 ms] Eagerly: scope を cancel
[ 964 ms] Eagerly: 上流 停止
[1014 ms] === Lazily ===
[1278 ms] Lazily: 購読者が collect 開始
[1285 ms] Lazily: 上流 開始
[1285 ms] Lazily: 上流 emit 1
[1285 ms] Lazily:   購読者 受信 1
[1385 ms] Lazily: 上流 emit 2
[1386 ms] Lazily:   購読者 受信 2
[1492 ms] Lazily: 上流 emit 3
[1501 ms] Lazily:   購読者 受信 3
[1529 ms] Lazily: 購読者が cancel
[1601 ms] Lazily: 上流 emit 4
[1702 ms] Lazily: 上流 emit 5
[1780 ms] Lazily: scope を cancel
[1780 ms] Lazily: 上流 停止
[1832 ms] === WhileSubscribed ===
[2156 ms] WhileSubscribed: 購読者が collect 開始
[2159 ms] WhileSubscribed: 上流 開始
[2162 ms] WhileSubscribed: 上流 emit 1
[2162 ms] WhileSubscribed:   購読者 受信 1
[2262 ms] WhileSubscribed: 上流 emit 2
[2263 ms] WhileSubscribed:   購読者 受信 2
[2363 ms] WhileSubscribed: 上流 emit 3
[2364 ms] WhileSubscribed:   購読者 受信 3
[2408 ms] WhileSubscribed: 購読者が cancel
[2409 ms] WhileSubscribed: 上流 停止
[2659 ms] WhileSubscribed: scope を cancel

Eagerly では購読者が来る前に 1・2・3 が流れ、購読者は replay = 1 の分の 3 から受け取りました。WhileSubscribed では購読をやめた 1 ミリ秒後に 上流 停止 が出ています。公式の API ドキュメントによると、WhileSubscribed には stopTimeoutMillis（最後の購読者がいなくなってから上流を止めるまでの待ち時間）と replayExpirationMillis（上流を止めてから replay 用の値を消すまでの時間）を渡せます。

stateIn は初期値を渡す

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val scope = CoroutineScope(coroutineContext + Job())
    val source = flow {
        println("上流 開始")
        for (i in 1..3) { delay(100); emit(i) }
    }

    val state: StateFlow<Int> = source.stateIn(scope, SharingStarted.Eagerly, initialValue = 0)
    println("直後の value = ${state.value}")
    delay(250)
    println("250ms 後の value = ${state.value}")

    val first: StateFlow<Int> = source.stateIn(scope)   // 最初の値が届くまで待つ suspend 版
    println("suspend 版の value = ${first.value}")
    scope.cancel()
}

直後の value = 0
上流 開始
250ms 後の value = 2
上流 開始
suspend 版の value = 1

stateIn(scope, started, initialValue) は、上流の最初の値が届くまで initialValue を持ちます。引数が scope だけの stateIn(scope) は suspend 関数で、最初の値が届くまで待ってから StateFlow を返すので、初期値を渡さずに済みます。

つまずきやすい点のチェックリスト

よくある質問

Q. StateFlow と SharedFlow はどちらを使えばよいですか？ A. 「今の値」をいつでも読めて、あとから来た購読者にも最新値を渡したいなら StateFlow です。同じ内容でも毎回届けたい通知や、1件ずつ確実に処理したい出来事には SharedFlow が合います。

Q. Flow とコルーチンの suspend 関数はどう使い分けますか？ A. 結果が1つなら suspend 関数、時間をおいて複数の値が届くなら Flow です。Flow から値を1つだけ取り出すときは first() が使えます。

Q. MutableStateFlow と StateFlow の違いは何ですか？ A. MutableStateFlow は value に代入できる型、StateFlow は読み取り専用の型です。クラスの中で MutableStateFlow を private に持ち、外には asStateFlow() で StateFlow として公開すると、値を書き換える場所を1か所にまとめられます。SharedFlow も asSharedFlow() で同じ形にできます。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

class Counter {
    private val _count = MutableStateFlow(0)
    val count: StateFlow<Int> = _count.asStateFlow()   // 外には読み取り専用で公開

    private val _events = MutableSharedFlow<String>(replay = 1)
    val events: SharedFlow<String> = _events.asSharedFlow()

    fun increment() {
        _count.update { it + 1 }                        // 今の値をもとに更新
        println("tryEmit = " + _events.tryEmit("count が ${_count.value} になりました"))
    }
}

fun main() = runBlocking {
    val c = Counter()
    c.increment()
    c.increment()
    println("count.value = ${c.count.value}")
    println("events.replayCache = ${c.events.replayCache}")
    println("first = ${c.events.first()}")
}

tryEmit = true
tryEmit = true
count.value = 2
events.replayCache = [count が 2 になりました]
first = count が 2 になりました

update { } は今の値をもとに次の値を決める書き方、tryEmit は suspend せずに値を入れて、入ったかどうかを Boolean で返す関数です。

Q. collect を止めるにはどうしますか？ A. collect を呼んだコルーチンの Job を cancel() するか、launchIn に渡した scope を cancel します。決まった個数で終えたいときは take(n) や first() を使うと、受け取った時点で自然に終わります。

まとめ

• flow { } はコールドフロー。collect のたびにブロックが最初から実行され、あとから collect しても全部受け取れる
• StateFlow / SharedFlow はホットフロー。あとから来た購読者に届くのは replay の分だけ（StateFlow は最新の1個）
• StateFlow は初期値が必須で、同じ値を間引き、value で今の値を読める。SharedFlow は同じ値も毎回届ける
• 演算子は「Flow を返すもの」と「終端演算子」に分かれる。終端演算子を呼んで初めて Flow が動く
• flowOn は上流のスレッドだけを変え、catch は上流の例外だけを受ける
• ホットフローの collect は終わらない。次の行へ進みたいときは launch で囲むか onEach { }.launchIn(scope) と書く
• stateIn / shareIn の started は、Eagerly（すぐ開始）・Lazily（最初の購読で開始）・WhileSubscribed（購読者がいる間だけ）から選ぶ

参考・出典

• Kotlin 公式ドキュメント「Asynchronous flow」: https://kotlinlang.org/docs/coroutines-flow.html
• kotlinx.coroutines API「StateFlow」: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-state-flow/
• kotlinx.coroutines API「SharingStarted」: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-sharing-started/
• kotlinx.coroutines README（依存の書き方）: https://github.com/Kotlin/kotlinx.coroutines/blob/master/README.md

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。