Mori Atsushi

kotlin coroutines 1.3.6 にて、StateFlowというものが導入されました。

状態管理のために用いられる型で、将来的にConflatedBroadcastChannelから置き換わるとも言われています。

今回は、ドキュメントを詳しく見つつ、実際にコードを動かして特徴について見ていきたいと思います。

- 2020/11/15追記 -

この記事はSharedFlowがリリースされる前に書かれています。

SharedFlowとの比較等をしている、こちらの記事も合わせて確認してください。

StateFlow

今回は、このドキュメントを上から順番に読んでいきます。

まず、定義はこうなっています。

@ExperimentalCoroutinesApi
interface StateFlow<out T> : Flow<T> {
  public val value: T
}

ExperimentalCoroutinesApi というのは、まだ実験段階のAPIで将来変更や廃止される可能性がありますよ、という意味です。

そしてStateFlowはFlowを継承しているということがわかります。

genericsのout という修飾子は、共変性を表します。(リファレンス)

MutableStateFlow

そして、StateFlowを作るためには、 MutableStateFlow というものを使います。

定義はこうです。

@ExperimentalCoroutinesApi
interface MutableStateFlow<T> : StateFlow<T> {
    public override var value: T
}

valだったvalueがvarになっており、outは外れて不変であることがわかりますね。

ちょうど、ListとMutableListのような関係であることがわかると思います。

公式ドキュメントに乗っている使い方はこうです。

class CounterModel {
    private val _counter = MutableStateFlow(0)
    val counter: StateFlow<Int> get() = _counter

    fun inc() {
        _counter.value++
    }
}

LiveDataを使ったことがある人であれば、使い方はすぐわかりますね。

Strong equality-based conflation

さて、ドキュメントに書かれているこの言葉の意味はどういうことでしょう？

簡単に言うと、StateFlowは distinctUntilChanged と同じ動作をし、同じ値は流れません。

これはLiveDataやConflatedBroadcastChannelとも違う特徴ですね。

試しに、こんなコードを書いて実行してみます。

runBlocking {
    val counter = MutableStateFlow(0)
    val job = launch {
        counter.collect {
            println(it)
        }
    }
    delay(100)
    counter.value = 0
    delay(100)
    counter.value = 1
    delay(100)
    counter.value = 1
    delay(100)
    job.cancel()
}

0
1

同じデータは流れてこないことがわかると思います。

ちなみに、collectは非同期に実行されるので、delayを入れないと何も出力されません。

runBlocking {
    val counter = MutableStateFlow(0)
    val job = launch {
        counter.collect {
            println(it)
        }
    }
    counter.value = 0
    counter.value = 1
    counter.value = 1
    job.cancel()
}

// 出力なし

ここから、イベント等を使う場合はBroadcastChannelを引き続き使ってください。

また、ここでいう同じデータ、というのが Any.equals によって決まります。

すなわち、 こんな感じのdata classを用意すると、このように動作します。

data class User(
    val id: Long,
    val name: String 
) {
    override fun equals(other: Any?): Boolean {
        return other is User && this.id == other.id
    }
}

runBlocking {
    val user = MutableStateFlow(User(0, "taro"))
    val job = launch {
        user.collect {
            println(it)
        }
    }
    delay(100)
    user.value = User(1, "jiro")
    delay(100)
    user.value = User(1, "jiro2")
    delay(100)
    job.cancel()
}

User(id=0, name=taro)
User(id=1, name=jiro)

idが変わった際のみデータが流れています。

このあたりは少し注意が必要かもしれません。

StateFlow vs ConflatedBroadcastChannel

次の章はConflatedBroadcastChannelとの比較です。

順番に見ていきます。

StateFlow is simpler, because it does not have to implement all the Channel APIs, which allows for faster, garbage-free implementation, unlike ConflatedBroadcastChannel implementation that allocates objects on each emitted value.

ConflatedBroadcastChannelはChannelのメソッドをすべてoverrideしてるため、複雑だったが、StateFlowはそうでないのでシンプルだ、と書かれています。

たとえば、ConflatedBroadcastChannelのofferは常にtrueを返し、実際はsendと動作が変わらない点等が挙げられると思います。

このあたりの複雑さを取り除いたという点では、StateFlowは優れていそうです。

StateFlow always has a value which can be safely read at any time via value property. Unlike ConflatedBroadcastChannel, there is no way to create a state flow without a value.

ConflatedBroadcastChannelは値なしで初期化することができました。

初期化されずにvalueを取得しようとすると、IllegalStateException が吐かれます。

val counter = ConflatedBroadcastChannel<Int>()
val value = counter.value // java.lang.IllegalStateException: No value

そのため、安全に値を取得するためには、valueOrNullを使う必要がありました。

一方、StateFlowは初期化時に初期値が必須なため、安全にvalue で値を取得されます。

初期値を入れたくない場合は、nullableなStateFlowにするか、sealed class等で初期値を定義する必要がありそうです。

StateFlow has a clear separation into a read-only StateFlow interface and a MutableStateFlow .

上記でも書いたとおり、StateFlowとMutableStateFlowでクラスが分かれていることが書かれています。

外部に公開する際に、必要な型で公開すれば良いので、これは便利そうです。

StateFlow conflation is based on equality like distinctUntilChanged operator, unlike conflation in ConflatedBroadcastChannel that is based on reference identity.

上記でも書いたとおり、ConflatedBroadcastChannelと違い等価な値は流れて来ないことが示されています。

StateFlow cannot be currently closed like ConflatedBroadcastChannel and can never represent a failure. This feature might be added in the future if enough compelling use-cases are found.

StateFlowはチャンネルではないので、closeすることが 現在は できません。

そのため、closeした後に値を取得しようとして失敗する、という以下ような現象は起こらなくなります。

val channel = ConflatedBroadcastChannel(1)
channel.offer(2)
channel.close()
val value = channel.value // java.lang.IllegalStateException: Channel was closed

しかし、これは将来的に変更される可能性があるので、注意が必要です。

以上の点から、状態を管理するのであれば、ConflatedBroadcastChannelよりもStateFlowを使ったほうがよりシンプルで適切であることがわかります。

Concurrency

StateFlowはスレッドセーフである、と書かれています。

このように別のスレッドで同時にインクリメントしても、問題なく2まで加算されることを示しています。

0, 1は出力されないことにも注目ですね。

runBlocking {
    val counter = MutableStateFlow(0)
    val job = launch {
        counter.collect {
            println(it)
        }
    }
    val deferred1 = async(Dispatchers.Main) {
        counter.value += 1
    }
    val deferred2 = async(Dispatchers.Default) {
        counter.value += 1
    }
    deferred1.await()
    deferred2.await()
    delay(100)
    job.cancel()
}

2

Operator fusion

以下のoperatorはStateFlowで使えません。

• flowOn
• conflate
• buffer with CONFLATED or RENDEZVOUS capacity
• distinctUntilChanged

StateFlowに対して直接使おうとすると、コンパイル時にエラーになります。

val counter = MutableStateFlow(0)
val flow = counter.distinctUntilChanged() // Using 'distinctUntilChanged(): Flow<T>' is an error. Applying distinctUntilChanged operator to StateFlow has no effect. See StateFlow documentation on Operator Fusion.

ちなみにこのようにFlowにキャストした後だと問題なく実行できますが、streamに変化を与えないため、無駄なのでやめておきましょう。

val counter = MutableStateFlow(0)
val flow = (counter as Flow<Int>).distinctUntilChanged()

Implementation notes

新しいコレクターの追加にはO(1)のコストがかかり、値の更新にはO(N)のコストがかかるというように書かれています。

試してませんが、むやみにコレクターを追加するのは危険かもしれません。

Not stable for inheritance

StateFlowのinterfaceはまだメソッド等が追加される可能性があるため、継承して使わないでね、と書かれています。

まとめ

StateFlowは以下のような特徴があることがわかりました。

• 同じ値は流れてこない
• collectは非同期で実行される
• ConflatedBroadcastChannelよりsimpleである
• 初期値が必須
• スレッドセーフである
• 使えないFlowのoperatorがある

Flowはcold streamである、という説明が崩れる大きな変更ですが、いくつかのメリットも見つかったと思います。

今後はConflatedBroadcastChannelよりはこちらのほうが使われるのではないかなと思います。

一方で、使えないoperatorがあったり、いまいち美しくないなぁと思う点もいくつかありました。

また、通常のFlowからStateFlowへの変換等も提供されていません。

今後もkotlin coroutines flowの動きには注視していきたいと思います。

- 2021/01/31追記-

Kotlin Coroutinesの解説本をZennにて販売しています。より詳しく学びたい方は、こちらも合わせて確認してみて下さい。

詳解 Kotlin Coroutines [2021] | Zenn