Android: Flow vs RxJava vs LiveData

這麼多 reactive stream 是要用哪一個？現在還多一個 Compose State

Ray Yuan Liou

9 min readFeb 27, 2024

New reactive stream every two years! — New reactive stream every two years???

Flow

相關名詞

producer: 推送數值
consumer: 收集消化 producer 製造的數值

Flow 原生支援 back pressure （像是 RxJava 2+），可以處理 producer 短時間內傳送大量資料。

Flow 本質是一種 Cold Flow，只有當有其他人 Collect (RxJava 叫做 subscribe)，他才會開始做事。 (這邊跟 RxJava 的 Observable /Single 等 reactive stream ... 類似）

例如，以下實作是一個倒數計時的 Flow

// producer
val countDownFlow = flow<Int> {
    val startValue = 10
    var currentValue = startValue
    emit(startValue)
    while (currentValue > 0) {
        delay(1000L) // stop a second
        currentValue--
        emit(curretValue)
    }
}

我們有多種不同的 collect 方式，例如：

// consumers
// assume we're in a ViewModel

private fun collectFlow() {
    viewModelScope.launch {
        countDownFlow.collect { time ->
            println("time is $time)
        }
    }
}

也可以這樣：

private fun collectFlow() {
    countDownFlow.onEach { time ->
        println("time is $time)
    }.launchIn(viewModelScope)
}

onEach 有點像 RxJava doOnNext ，在每次收到資料時都會被觸發。

Cancel Handling

.onCompletion { error ->
    if (error is CancellationException) {
        Log.i("MyCoolViewModel", "fetchCoolBanners is canceled")
    }
}

Kotlin Flow operator

reduce vs fold

fold 就是「有預設值」的 reduce，行為基本上都是累加前一個累積數值。如果給一個空 Collection 給它 fold，他會直接回傳預設值。

flatMapConcat vs flatMapMerge

其實就是 RxJava flatMap 和 concatMap 的差別，flatMapConcat 有順序性而 flatMapMerge 則無，會一次打出去。Kotlin 官方這邊建議用 flatMapConcat 就好，少用 flatMapMerge 。

delay

上游 Flow emit 資料後，會等待下游 collect 結束後再往下走，換句話說 collect 裡有 delay 他就會等多久。 (這即是 back pressure 的體現：一般使用下的 Flow producer 和 consumer 的 code block 是跑在同一個 coroutines 裡面的！)

buffer vs conflate

這時如果不想讓上游 Flow 等待，可以用 buffer 這個 operator。如此一來 collect 的 scope 會跑在不同的 coroutines，上游也不會等下游 collect 做完，會持續發射數值。

但如果 buffer 的 size 不足，可能會導致 overflow，需要慎用。

conflate 也會繼續走下去，不過如果在發射新的數值後，下遊來不及 collect 就會直接被跳過。透過 conflate 蒐集的 flow 會在上一個數值完成 collect 之後，去取得最新被射出的數值。（也就是最後一個 emit 的數值)

根據這兩個 operator 的作用，我們可以再次體現到一般使用下的 Flow producer 和 consumer 的 code block 就是跑在同一個 coroutines 裡面的！

collectLatest

如果有新的資料吐出來，會放棄目前正在處理的資料，去 collect 最新的資料。

Hot Flows

與 Flow 本體不同，就算沒有人 collect，也可以持續的送出資料。

相對於 Cold Flow，拿個比方來說：

Cold Flow 就像是 CD/DVD，需要放到播放器裡才會播放內容；Hot Flow 則像是廣播/電視頻道，無論有沒有人收聽都會持續放送。

或 Cold Flow 像 YouTube 裡頭的影片，不點進去不會播；Hot Flow 則像是直播，無論有沒有人收看都會持續播。

StateFlow

基本上是失去察覺 Lifecycle 能力的 LiveData，會存最後一個 Data，在 collect 的瞬間也會吐上一個最近的 Data 給你。(就像 LiveData！)
例如，我們設定讓一個 StateFlow 的 ViewEvent 顯示一個 Toast，當裝置旋轉時，會再顯示出來一次。因為這邊 Activity 重新 recreate 後重新 collect 一次，stateFlow 又暴露出上一個值。

需要注意的是，stateFlow 正常使用且 Activity 無 recreate() 的情況下，他是不會吐相同的資料的。

any instance of StateFlow already behaves as if distinctUntilChanged operator is applied to it.

此外，StateFlow 需要給一個預設值。

因為他不知道 App 目前的狀態，所以在背景也是 active 的 (LiveData 則否)，但可以透過方法讓他表現的很像 LiveData。

以下是類似的 Stream：

這些 Stream 行為類似，可以作為類比

可以這樣 collect StateFlow，讓他行為像是 LiveData：

lifecycleScope.launch {
    // you can change to any lifecycle state you want
    // we set to STARTED here to emulated behaviors as a LiveData
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        viewModel.stateFlow.collectLatest {
            // collect to your data here
        }
    }
}

要注意的是，官方已經不建議這樣 collect： lifecycleScope.launchWhenStarted ，連文件都移除了。

SharedFlow

用來做 One-Time Event，不需要給他預設值。

可以透過建構子的 replay 參數來設定 Cache，代表可以將同一個值 Keep 幾次。而且 SharedFlow 並沒有 distinctUntilChanged 的特性，可以重複吐一樣的資料。

它需要在 coroutines 裡頭 emit value。

可以參考我以前的文章來看看怎麼實作 SingleLiveData

技術選擇

LiveData vs StateFlow

兩者可以做到幾乎相同的事情

choose LiveData

專案使用 Java

choose StateFlow

更好寫 unit test
沒有與 Android Context 綁太深，以及可以在 Kotlin coroutines 的基礎上做測試
更強大的 flow operators

Suspend function vs Flow vs StateFlow vs SharedFlow

choose Suspend function

非同步工作只需要回傳一個資料

choose Flow

非同步工作可以連續回傳資料，或回傳多個資料

choose StateFlow

傳送 State 到你的 UI，當作是 LiveData 的替代品

choose SharedFlow

傳送一次性事件到你的 UI，當作是 SingleLiveEvent 的替代品

StateFlow/SharedFlow vs Compose State

本比較基於 UI Layer (VM <-> View) 的溝通上

choose StateFlow/SharedFlow

專案 UI 建立於 XML
compose 並不支援 StateFlow/SharedFlow，需要透過 collectAsState() 來監聽。會失去原來 SharedFlow 的特性。（變得不是一次性 Event，他會變成 Compose State）所以不推薦 compose 搭配 SharedFlow 使用
Compose 已經有自己的 State holder，compose 搭配 StateFlow 使用需要轉換。

2024/02/29 Update: Perry Lu 提到的 StateFlow 留言也很棒，記得看留言區!

choose Compose State

專案 UI 建立於 Compose

Compose State 是唯一的 State holder

compose 依賴於 compose state 裡頭的狀態變化來做 recomposition (重繪)

在 Compose 裡頭 collect SharedFlow

用 LaunchedEffect 來做。

setContent {
    demoAppTheme {
        val viewModel = viewModel<DemoViewModel>()
        val context = LocalContext.current
        LaunchedEffect(key1 = true) {
            viewModel.sharedFlow.collect { viewEffect ->
                // collect one-time event here
                if (viewEffect is ShowSnackBar) {
                    SnackBar.make(context, viewEffect.text, SnackBar.LONG).show()
                }
            }
        }
        
        Box {
            Text("Hello world", modifier.centerVertical()) 
        }
    }
}

如何測試 Flow？

可以用 turbine 用來對 Flow 做 Unit Test