Kotlin Flows

本文包含的內容:

Flow是什么, 基本概念和用法.
Flow的不同型別, StateFlow和SharedFlow比較.
Flow在Android中的使用
- 安全收集.
- 運算子stateIn, shareIn的用法和區別.

本文被收錄在集合中: https://github.com/mengdd/KotlinTutorials

Coroutines Flow Basics

Flow是什么

Flow可以按順序發送多個值, 概念上是一個資料流, 發射的值必須是同一個型別.
Flow使用suspend方法來生產/消費值, 資料流可以做異步計算.

幾個基本知識點:

創建flow: 通過flow builders
Flow資料流通過emit()來發射元素.
可以通過各種運算子對flow的資料進行處理. 注意中間的運算子都不會觸發flow的資料發送.
Flow默認是cold flow, 即需要通過被觀察才能激活, 最常用的運算子是collect().
Flow的CoroutineContext, 不指定的情況下是collect()的CoroutineContext, 如果想要更改, 用flowOn
改之前的.

關于Flow的基本用法, 19年底寫的這篇coroutines flow in Android可以溫故知新.

Flow的運算子

一個Flow運算子的可視化小網站: FlowMarbles.

Flow的不同型別

SharedFlow and StateFlow

應用程式里比較常用的型別是SharedFlow和StateFlow.
Android官方有一篇專門的檔案來介紹二者: StateFlow and SharedFlow
StateFlow繼承于SharedFlow, SharedFlow繼承于Flow.

基本關系如下:

kotlin-flows

Flow
基類. Cold.
Flow的兩大特性: Context preservation; Exception transparency.
SharedFlow
繼承Flow, 是一種hot flow, 所有collectors共享它的值, 永不終止, 是一種廣播的方式.
一個shared flow上的活躍collector被叫作subscriber.

在sharedFlow上的collect call永遠不會正常complete, 還有Flow.launchIn.
可以配置replay and buffer overflow strategy.

如果subscriber suspend了, sharedflow會suspend這個stream, buffer這個要發射的元素, 等待subscriber resume.
Because onBufferOverflow is set with BufferOverflow.SUSPEND, the flow will suspend until it can deliver the event to all subscribers.

默認引數:

public fun <T> MutableSharedFlow(
    replay: Int = 0,
    extraBufferCapacity: Int = 0,
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND
)

total buffer是: replay + extraBufferCapacity.
如果total buffer是0, 那么onBufferOverflow只能是onBufferOverflow = BufferOverflow.SUSPEND.

關于reply和buffer, 這個文章
有詳細的解釋, 并且配有動圖.

StateFlow
繼承SharedFlow, hot flow, 和是否有collector收集無關, 永不complete.

可以通過value屬性訪問當前值.
有conflated特性, 會跳過太快的更新, 永遠回傳最新值.
Strong equality-based conflation: 會通過equals()來判斷值是否發生改變, 如果沒有改變, 則不會通知collector.
因為conflated的特性, StateFlow賦值的時候要注意使用不可變的值.

cold vs hot

cold stream 可以重復收集, 每次收集, 會對每一個收集者單獨開啟一次.
hot stream 永遠發射不同的值, 和是否有人收集無關, 永遠不會終止.

sharedIn
可以把cold flow轉成hot的SharedFlow.
stateIn
可以把cold flow轉成hot的StateFlow.

StateFlow vs SharedFlow

共性:

StateFlow和SharedFlow永遠都不會停止. 不能指望它們的onCompletionCallback.

不同點:

StateFlow可以通過value屬性讀到最新的值, 但SharedFlow卻不行.
StateFlow是conflated: 如果新的值和舊的值一樣, 不會傳播.
SharedFlow需要合理設定buffer和replay策略.

互相轉換:
SharedFlow用了distinctUntilChanged以后變成StateFlow.

// MutableStateFlow(initialValue) is a shared flow with the following parameters:
val shared = MutableSharedFlow(
    replay = 1,
    onBufferOverflow = BufferOverflow.DROP_OLDEST
)
shared.tryEmit(initialValue) // emit the initial value
val state = shared.distinctUntilChanged() // get StateFlow-like behavior

RxJava的等價替代:

PublishSubject -> SharedFlow.
BehaviorSubject -> StateFlow.

Use Flow in Android

發送事件(Event或Effects): SharedFlow

因為SharedFlow沒有conflated特性, 所以適合發送事件, 即便值變化得快也是每個都發送.

private val _sharedViewEffects = MutableSharedFlow<SharedViewEffects>() // 1
val sharedViewEffects = _sharedViewEffects.asSharedFlow() // 2

這里用了asSharedFlow來創建一個ReadonlySharedFlow.

SharedFlow發射元素有兩個方法:

emit: suspend方法.
tryEmit: 非suspend方法.

因為tryEmit是非suspend的, 適用于有buffer的情況.

保存暴露UI狀態: StateFlow

StateFlow是一個state-holder, 可以通過value讀到當前狀態值.
一般會有一個MutableStateFlow型別的Backing property.

StateFlow是hot的, collect并不會觸發producer code.
當有新的consumer時, 新的consumer會接到上次的狀態和后續的狀態.

使用StateFlow時, 發射新元素只需要賦值:

mutableState.value = https://www.cnblogs.com/mengdd/archive/2021/08/30/newState

注意這里新值和舊的值要equals判斷不相等才能發射出去.

StateFlow vs LiveData

StateFlow和LiveData很像.

StateFlow和LiveData的相同點:

永遠有一個值.
只有一個值.
支持多個觀察者.
在訂閱的瞬間, replay最新的值.

有一點點不同:

StateFlow需要一個初始值.
LiveData會自動解綁, flow要達到相同效果, collect要在Lifecycle.repeatOnLifecycle里.

Flow的安全收集

關于收集Flow的方法, 主要還是關注一下生命周期的問題, 因為SharedFlow和StateFlow都是hot的.
在這個文章里有詳細的討論: A safer way to collect flows from Android UIs

在UI層收集的時候注意要用repeatOnLifecycle:

class LatestNewsActivity : AppCompatActivity() {
    private val latestNewsViewModel = // getViewModel()

    override fun onCreate(savedInstanceState: Bundle?) {
        //...
        // Start a coroutine in the lifecycle scope
        lifecycleScope.launch {
            // repeatOnLifecycle launches the block in a new coroutine every time the
            // lifecycle is in the STARTED state (or above) and cancels it when it's STOPPED.
            repeatOnLifecycle(Lifecycle.State.STARTED) {
                // Trigger the flow and start listening for values.
                // Note that this happens when lifecycle is STARTED and stops
                // collecting when the lifecycle is STOPPED
                latestNewsViewModel.uiState.collect { uiState ->
                    // New value received
                    when (uiState) {
                        is LatestNewsUiState.Success -> showFavoriteNews(uiState.news)
                        is LatestNewsUiState.Error -> showError(uiState.exception)
                    }
                }
            }
        }
    }
}

這個文章里有個擴展方法也挺好的:

class FlowObserver<T> (
    lifecycleOwner: LifecycleOwner,
    private val flow: Flow<T>,
    private val collector: suspend (T) -> Unit
) {

    private var job: Job? = null

    init {
        lifecycleOwner.lifecycle.addObserver(LifecycleEventObserver {
                source: LifecycleOwner, event: Lifecycle.Event ->
            when (event) {
                Lifecycle.Event.ON_START -> {
                    job = source.lifecycleScope.launch {
                        flow.collect { collector(it) }
                    }
                }
                Lifecycle.Event.ON_STOP -> {
                    job?.cancel()
                    job = null
                }
                else -> { }
            }
        })
    }
}


inline fun <reified T> Flow<T>.observeOnLifecycle(
    lifecycleOwner: LifecycleOwner,
    noinline collector: suspend (T) -> Unit
) = FlowObserver(lifecycleOwner, this, collector)

inline fun <reified T> Flow<T>.observeInLifecycle(
    lifecycleOwner: LifecycleOwner
) = FlowObserver(lifecycleOwner, this, {})

看了一下官方的repeatOnLifecycle其實大概也是這個意思:

public suspend fun Lifecycle.repeatOnLifecycle(
    state: Lifecycle.State,
    block: suspend CoroutineScope.() -> Unit
) {
    require(state !== Lifecycle.State.INITIALIZED) {
        "repeatOnLifecycle cannot start work with the INITIALIZED lifecycle state."
    }

    if (currentState === Lifecycle.State.DESTROYED) {
        return
    }

    // This scope is required to preserve context before we move to Dispatchers.Main
    coroutineScope {
        withContext(Dispatchers.Main.immediate) {
            // Check the current state of the lifecycle as the previous check is not guaranteed
            // to be done on the main thread.
            if (currentState === Lifecycle.State.DESTROYED) return@withContext

            // Instance of the running repeating coroutine
            var launchedJob: Job? = null

            // Registered observer
            var observer: LifecycleEventObserver? = null
            try {
                // Suspend the coroutine until the lifecycle is destroyed or
                // the coroutine is cancelled
                suspendCancellableCoroutine<Unit> { cont ->
                    // Lifecycle observers that executes `block` when the lifecycle reaches certain state, and
                    // cancels when it falls below that state.
                    val startWorkEvent = Lifecycle.Event.upTo(state)
                    val cancelWorkEvent = Lifecycle.Event.downFrom(state)
                    val mutex = Mutex()
                    observer = LifecycleEventObserver { _, event ->
                        if (event == startWorkEvent) {
                            // Launch the repeating work preserving the calling context
                            launchedJob = [email protected] {
                                // Mutex makes invocations run serially,
                                // coroutineScope ensures all child coroutines finish
                                mutex.withLock {
                                    coroutineScope {
                                        block()
                                    }
                                }
                            }
                            return@LifecycleEventObserver
                        }
                        if (event == cancelWorkEvent) {
                            launchedJob?.cancel()
                            launchedJob = null
                        }
                        if (event == Lifecycle.Event.ON_DESTROY) {
                            cont.resume(Unit)
                        }
                    }
                    [email protected](observer as LifecycleEventObserver)
                }
            } finally {
                launchedJob?.cancel()
                observer?.let {
                    [email protected](it)
                }
            }
        }
    }
}

既然官方已經推出了, 我們就用官方的repeatOnLifecycle方法吧.

`shareIn`和`stateIn`

前面提過這兩個運算子是用來做flow轉換的:

sharedIn
可以把cold flow轉成hot的SharedFlow.
stateIn
可以把cold flow轉成hot的StateFlow.

shareIn可以保證只有一個資料源被創造, 并且被所有collectors收集.
比如:

class LocationRepository(
    private val locationDataSource: LocationDataSource,
    private val externalScope: CoroutineScope
) {
    val locations: Flow<Location> = 
        locationDataSource.locationsSource.shareIn(externalScope, WhileSubscribed())
}

WhileSubscribed這個策略是說, 當無人觀測時, 上游的flow就被取消.

實際使用時可以用WhileSubscribed(5000), 讓上游的flow即便在無人觀測的情況下, 也能繼續保持5秒.
這樣可以在某些情況(比如旋轉螢屏)時避免重建上游資源, 適用于上游資源創建起來很expensive的情況.

如果我們的需求是, 永遠保持一個最新的cache值.


class LocationRepository(
    private val locationDataSource: LocationDataSource,
    private val externalScope: CoroutineScope
) {
    val locations: Flow<Location> = 
        locationDataSource.locationsSource.stateIn(externalScope, WhileSubscribed(), EmptyLocation)
}

Flow.stateIn將會快取最后一個值, 并且有新的collector時, 將這個最新值傳給它.

`shareIn`, `stateIn`使用注意事項

永遠不要在方法里面呼叫shareIn和stateIn, 因為方法每次被呼叫, 它們都會創建新的流.
這些流沒有被復用, 會存在記憶體里面, 直到scope被取消或者沒有參考時被GC.

推薦的使用方式是在property上用:

class UserRepository(
    private val userLocalDataSource: UserLocalDataSource,
    private val externalScope: CoroutineScope
) {
    // DO NOT USE shareIn or stateIn in a function like this.
    // It creates a new SharedFlow/StateFlow per invocation which is not reused!
    fun getUser(): Flow<User> =
        userLocalDataSource.getUser()
            .shareIn(externalScope, WhileSubscribed())    

    // DO USE shareIn or stateIn in a property
    val user: Flow<User> = 
        userLocalDataSource.getUser().shareIn(externalScope, WhileSubscribed())
}

StateFlow使用總結

從ViewModel暴露資料到UI, 用StateFlow的兩種方式:

暴露一個StateFlow屬性, 用WhileSubscribed加上一個timeout.

class MyViewModel(...) : ViewModel() {
    val result = userId.mapLatest { newUserId ->
        repository.observeItem(newUserId)
    }.stateIn(
        scope = viewModelScope, 
        started = WhileSubscribed(5000), 
        initialValue = https://www.cnblogs.com/mengdd/archive/2021/08/30/Result.Loading
    )
}

用repeatOnLifecycle收集.

onCreateView(...) {
    viewLifecycleOwner.lifecycleScope.launch {
        viewLifecycleOwner.lifecycle.repeatOnLifecycle(STARTED) {
            myViewModel.myUiState.collect { ... }
        }
    }
}

其他的組合都會保持上游的活躍, 浪費資源:

用WhileSubscribed暴露屬性, 在lifecycleScope.launch/launchWhenX里收集.
通過Lazily/Eagerly暴露, 用repeatOnLifecycle收集.

References

Kotlin flows on Android
StateFlow and SharedFlow
A safer way to collect flows from Android UIs
Things to know about Flow’s shareIn and stateIn operators
Shared flows, broadcast channels
Kotlin SharedFlow or: How I learned to stop using RxJava and love the Flow
Migrating from LiveData to Kotlin’s Flow
Substituting Android’s LiveData: StateFlow or SharedFlow?
Learning State & Shared Flows with Unit Tests
Reactive Streams on Kotlin: SharedFlow and StateFlow
Reading Coroutine official guide thoroughly — Part 0

作者: 圣騎士Wind
出處: 博客園: 圣騎士Wind
Github: https://github.com/mengdd
微信公眾號: 圣騎士Wind

轉載請註明出處，本文鏈接：https://www.uj5u.com/yidong/296457.html

標籤：其他

上一篇：百度地圖開發-實作離線地圖功能 05

下一篇：百度地圖開發-與地圖的互動功能 06

[Kotlin Tutorials 19] Kotlin Flows, SharedFlow and StateFlow in Android