Kotlin Flow封裝類SharedFlow StateFlow LiveData使用對(duì)比

更新時(shí)間：2022年08月31日 14:47:43 作者：newki

這篇文章主要為大家介紹了Kotlin Flow封裝類SharedFlow StateFlow LiveData使用對(duì)比，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

Kotlin中SharedFlow的使用 VS StateFlow

SharedFlow 是繼承于 Flow ,同時(shí)它是 StateFlow 的父類,它們都是是熱流，先說一下冷流與熱流的概念。

冷流 :只有訂閱者訂閱時(shí)，才開始執(zhí)行發(fā)射數(shù)據(jù)流的代碼。并且冷流和訂閱者只能是一對(duì)一的關(guān)系，當(dāng)有多個(gè)不同的訂閱者時(shí)，消息是重新完整發(fā)送的。也就是說對(duì)冷流而言，有多個(gè)訂閱者的時(shí)候，他們各自的事件是獨(dú)立的。
熱流:無論有沒有訂閱者訂閱，事件始終都會(huì)發(fā)生。當(dāng) 熱流有多個(gè)訂閱者時(shí)，熱流與訂閱者們的關(guān)系是一對(duì)多的關(guān)系，可以與多個(gè)訂閱者共享信息。

SharedFlow的特點(diǎn)

SharedFlow沒有默認(rèn)值
SharedFlow可以保存舊的數(shù)據(jù)，根據(jù)配置可以將舊的數(shù)據(jù)回播給新的訂閱者
SharedFlow使用emit/tryEmit發(fā)射數(shù)據(jù)，StateFlow內(nèi)部其實(shí)都是調(diào)用的setValue。
SharedFlow會(huì)掛起直到所有的訂閱者處理完成。

為什么我先講的 StateFlow ，而不是SharedFlow，是因?yàn)?StateFlow 是繼承 SharedFlow 實(shí)現(xiàn)，是在其基礎(chǔ)的場(chǎng)景化實(shí)現(xiàn)，我們可以把 StateFlow 理解為是 SharedFlow 的 “青春版”。并不是它更輕量，而是它使用更簡(jiǎn)單。

我們舉例看看怎么使用 SharedFlow，看看它與 StateFlow的區(qū)別。

既然 StateFlow 是繼承 SharedFlow 實(shí)現(xiàn)，那么StateFlow

一、SharedFlow的使用

方式一，我們自己 new 出來

public fun <T> MutableSharedFlow(
    // 重放數(shù)據(jù)個(gè)數(shù)
    replay: Int = 0,
    // 額外緩存容量
    extraBufferCapacity: Int = 0,
    // 緩存溢出策略
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND
): MutableSharedFlow<T> {
    val bufferCapacity0 = replay + extraBufferCapacity
    val bufferCapacity = if (bufferCapacity0 < 0) Int.MAX_VALUE else bufferCapacity0 // coerce to MAX_VALUE on overflow
    return SharedFlowImpl(replay, bufferCapacity, onBufferOverflow)
}
public enum class BufferOverflow {
    // 掛起
    SUSPEND,
    // 丟棄最早的一個(gè)
    DROP_OLDEST,
    // 丟棄最近的一個(gè)
    DROP_LATEST
}

舉例說明

@HiltViewModel
class Demo4ViewModel @Inject constructor(
    val savedState: SavedStateHandle
) : BaseViewModel() {
    private val _sharedFlow = MutableSharedFlow<String>(replay = 1, onBufferOverflow = BufferOverflow.SUSPEND)
    val sharedFlow: SharedFlow<String> = _sharedFlow
    fun changeSearch(keyword: String) {
        _sharedFlow.tryEmit(keyword)
    }
}

在Activity中我們就可以像類似 LiveData 一樣的使用 SharedFlow

    private fun testflow() {
       mViewModel.changeSearch("key")
    }
    override fun startObserve() {
        mViewModel.sharedFlow.collect {
            YYLogUtils.w("value $it")
        }
    }

方式二，通過一個(gè) 冷流 Flow 轉(zhuǎn)換為 sharedFlow

class NewsRemoteDataSource(...,
    private val externalScope: CoroutineScope,
) {
    val latestNews: Flow<List<ArticleHeadline>> = flow {
        ...
    }.shareIn(
        externalScope,
        replay = 1,
        started = SharingStarted.WhileSubscribed() // 啟動(dòng)政策
    )
}

幾個(gè)重要參數(shù)的說明如下

scope 共享開始時(shí)所在的協(xié)程作用域范圍
started 控制共享的開始和結(jié)束的策略
replay 為0 代表不重放，也就是沒有粘性,為1 代表重放最新的一個(gè)數(shù)據(jù)

scope 和 replay 不需要過多解釋，主要介紹下 started: SharingStarted 啟動(dòng)策略，分為三種：

Eagerly（熱啟動(dòng)式）：立即啟動(dòng)數(shù)據(jù)流，并保持?jǐn)?shù)據(jù)流（直到 scope 指定的作用域結(jié)束）；

Lazily（懶啟動(dòng)式）：在首個(gè)訂閱者注冊(cè)時(shí)啟動(dòng)，并保持?jǐn)?shù)據(jù)流（直到 scope 指定的作用域結(jié)束）；

WhileSubscribed()：在首個(gè)訂閱者注冊(cè)時(shí)啟動(dòng)，并保持?jǐn)?shù)據(jù)流直到在最后一個(gè)訂閱者注銷時(shí)結(jié)束（或直到 scope 指定的作用域結(jié)束）。

使用示例：

        val sharedFlow = flowOf(1, 2, 3).shareIn(
            scope = lifecycleScope,
//            started = WhileSubscribed(5000, 1000),
//            started = Eagerly,
            started = Lazily,
            replay = 0
        )
        lifecycleScope.launch {
            sharedFlow.collect {
                YYLogUtils.w("shared-value $it")
            }
        }

打印結(jié)果：

創(chuàng)建的幾種方式基本和StateFlow類似，那么它們之間有什么區(qū)別？

二、SharedFlow、StateFlow、LiveData的對(duì)比

我們直接舉例，實(shí)現(xiàn) LiveData 的功能。我們看看 LiveData StateFlow SharedFlow 實(shí)現(xiàn)同樣的效果如何操作

@HiltViewModel
class Demo4ViewModel @Inject constructor(
    val savedState: SavedStateHandle
) : BaseViewModel() {
    private val _searchLD = MutableLiveData<String>()
    val searchLD: LiveData<String> = _searchLD
    private val _searchFlow = MutableStateFlow("")
    val searchFlow: StateFlow<String> = _searchFlow
    private val _sharedFlow = MutableSharedFlow<String>(replay = 1, onBufferOverflow = BufferOverflow.SUSPEND)
    val sharedFlow: SharedFlow<String> = _sharedFlow
    fun changeSearch(keyword: String) {
        _sharedFlow.tryEmit(keyword)
        _searchFlow.value = keyword
        _searchLD.value = keyword
    }
}

打印的結(jié)果：

可以看到 SharedFlow 通過設(shè)置之后是可以達(dá)到 LiveData 和 StateFlow 的效果的。

SharedFlow對(duì)比StateFlow的優(yōu)勢(shì)，不需要設(shè)置默認(rèn)值，沒有默認(rèn)值的發(fā)送。

SharedFlow對(duì)比StateFlow的劣勢(shì)，不能自由取值，這是致命的。

例如下面的代碼，StateFlow 我可以在代碼的任意地方取值，但是 SharedFlow 只能接收流，不能自由取值。

所以，我們一般才說 StateFlow 平替 LiveData，雖然 SharedFlow 可以通過參數(shù)的方式達(dá)到一部分 LiveData 的效果，但是痛點(diǎn)更明顯。

另外需要說明的是 StateFlow 與 SharedFlow 這么設(shè)置是去重的，也就是說如果點(diǎn)擊登錄按鈕之后登錄失敗報(bào)告密碼錯(cuò)誤，然后再次點(diǎn)擊登錄按鈕，就不會(huì)彈出吐司了。

這不符合我們的業(yè)務(wù)場(chǎng)景啊，如果按照 StateFlow 平替 LiveData 的原則，我們還需要改用 Channel 的方式才行（畢竟SharedFlow不能自由取值真的不適合這個(gè)場(chǎng)景）。

@HiltViewModel
class Demo4ViewModel @Inject constructor(
    val savedState: SavedStateHandle
) : BaseViewModel() {
    val channel = Channel<String>(Channel.CONFLATED)
    private val _searchLD = MutableLiveData<String>()
    val searchLD: LiveData<String> = _searchLD
    private val _searchFlow = MutableStateFlow("")
    val searchFlow: StateFlow<String> = _searchFlow
    private val _sharedFlow = MutableSharedFlow<String>(replay = 1, onBufferOverflow = BufferOverflow.SUSPEND)
    val sharedFlow: SharedFlow<String> = _sharedFlow
    fun changeSearch(keyword: String) {
        _sharedFlow.tryEmit(keyword)
        _searchFlow.value = keyword
        _searchLD.value = keyword
        channel.trySend(keyword)
    }
}

    private fun testflow() {
        mViewModel.changeSearch("1234")
    }
    override fun startObserve() {
        mViewModel.searchLD.observe(this) {
            YYLogUtils.w("value $it")
        }
        lifecycleScope.launch {
            mViewModel.sharedFlow.collect {
                YYLogUtils.w("shared-value1 $it")
            }
        }
        lifecycleScope.launch {
            mViewModel.channel.consumeAsFlow().collect {
                YYLogUtils.w("shared-value2 $it")
            }
        }
        lifecycleScope.launchWhenCreated {
            mViewModel.searchFlow.collect {
                YYLogUtils.w("state-value $it")
            }
        }
    }

我們加入了使用 Channel 的方式，前文我們講過 Channel 是協(xié)程中的通信通道，我們這邊發(fā)送那一邊轉(zhuǎn)為Flow來collect。打印結(jié)果如下：

好麻煩哦，這還不如LiveData呢，所以大家知道 StateFlow 與 LiveData 的優(yōu)缺點(diǎn)之后，按需選擇即可。

三、SharedFlow 的粘性設(shè)置與事件總線

可以看到雖然 SharedFlow 不能平替 LiveData ,但是它在事件的發(fā)送與接收相關(guān)的配置與使用到時(shí)得天獨(dú)厚，我們常用于事件總線的實(shí)現(xiàn)，例如SharedFlowBus，用于替代 EventBus

object FlowBus {
    private val busMap = mutableMapOf<String, EventBus<*>>()
    private val busStickMap = mutableMapOf<String, StickEventBus<*>>()
    @Synchronized
    fun <T> with(key: String): EventBus<T> {
        var eventBus = busMap[key]
        if (eventBus == null) {
            eventBus = EventBus<T>(key)
            busMap[key] = eventBus
        }
        return eventBus as EventBus<T>
    }
    @Synchronized
    fun <T> withStick(key: String): StickEventBus<T> {
        var eventBus = busStickMap[key]
        if (eventBus == null) {
            eventBus = StickEventBus<T>(key)
            busStickMap[key] = eventBus
        }
        return eventBus as StickEventBus<T>
    }
    //真正實(shí)現(xiàn)類
    open class EventBus<T>(private val key: String) : LifecycleObserver {
        //私有對(duì)象用于發(fā)送消息
        private val _events: MutableSharedFlow<T> by lazy {
            obtainEvent()
        }
        //暴露的公有對(duì)象用于接收消息
        val events = _events.asSharedFlow()
        open fun obtainEvent(): MutableSharedFlow<T> = MutableSharedFlow(0, 1, BufferOverflow.DROP_OLDEST)
        //主線程接收數(shù)據(jù)
        fun register(lifecycleOwner: LifecycleOwner, action: (t: T) -> Unit) {
            lifecycleOwner.lifecycle.addObserver(this)
            lifecycleOwner.lifecycleScope.launch {
                events.collect {
                    try {
                        action(it)
                    } catch (e: Exception) {
                        e.printStackTrace()
                        YYLogUtils.e("FlowBus - Error:$e")
                    }
                }
            }
        }
        //協(xié)程中發(fā)送數(shù)據(jù)
        suspend fun post(event: T) {
            _events.emit(event)
        }
        //主線程發(fā)送數(shù)據(jù)
        fun post(scope: CoroutineScope, event: T) {
            scope.launch {
                _events.emit(event)
            }
        }
        //自動(dòng)銷毀
        @OnLifecycleEvent(Lifecycle.Event.ON_DESTROY)
        fun onDestroy() {
            YYLogUtils.w("FlowBus - 自動(dòng)onDestroy")
            val subscriptCount = _events.subscriptionCount.value
            if (subscriptCount <= 0)
                busMap.remove(key)
        }
    }
    class StickEventBus<T>(key: String) : EventBus<T>(key) {
        override fun obtainEvent(): MutableSharedFlow<T> = MutableSharedFlow(1, 1, BufferOverflow.DROP_OLDEST)
    }
}

發(fā)送與接收消息

 // 主線程-發(fā)送消息
    FlowBus.with<String>("test-key-01").post(this@Demo11OneFragment2.lifecycleScope, "Test Flow Bus Message")

 // 接收消息
    FlowBus.with&lt;String&gt;("test-key-01").register(this) {
            LogUtils.w("收到FlowBus消息 - " + it)
        }

發(fā)送粘性消息
 FlowBus.withStick<String>("test-key-02").post(lifecycleScope, "Test Stick Message")

 // 接收粘性消息
FlowBus.withStick<String>("test-key-02").register(this){
        LogUtils.w("收到粘性消息：$it")
    }

看源碼就知道粘性的實(shí)現(xiàn)就得益于 SharedFlow 的構(gòu)造參數(shù)

replay的設(shè)置 ，代表重放的數(shù)據(jù)個(gè)數(shù)

replay 為0 代表不重放，也就是沒有粘性

replay 為1 代表重放最新的一個(gè)數(shù)據(jù)，后來的接收器能接受1個(gè)最新數(shù)據(jù)。

replay 為2 代表重放最新的兩個(gè)數(shù)據(jù)，后來的接收器能接受2個(gè)最新數(shù)據(jù)。

我們知道Flow的操作符有針對(duì)背壓的處理，那么 SharedFlow 內(nèi)部還對(duì)背壓做了快速處理。我們只需要通過參數(shù)快速設(shè)置即可實(shí)現(xiàn)。

extraBufferCapacity的設(shè)置，額外數(shù)據(jù)的緩存

當(dāng)上游事件發(fā)送過快，而消費(fèi)太慢的情況，這種情況下，就需要使用緩存池，把未消費(fèi)的數(shù)據(jù)存下來。

緩沖池容量 = replay + extraBufferCapacity

如果總量為 0 ，就 Int.MAX_VALUE

onBufferOverflow的設(shè)置

如果指定了有限的緩存容量，那么超過容量以后怎么辦？

BufferOverflow.SUSPEND ：超過就掛起，默認(rèn)實(shí)現(xiàn)

BufferOverflow.DROP_OLDEST : 丟棄最老的數(shù)據(jù)

BufferOverflow.DROP_LATEST : 丟棄最新的數(shù)據(jù)

總結(jié)

StateFlow 更加簡(jiǎn)便特定的場(chǎng)景使用，而 SharedFlow 更加的靈活，他們兩者的側(cè)重點(diǎn)也不同。

SharedFlow 基于緩存的處理可以實(shí)現(xiàn)一些特定的需求，如當(dāng)發(fā)生訂閱時(shí)，我需要將過去已經(jīng)更新的N個(gè)值，同步給新的訂閱者。比如有多個(gè)新的訂閱者都想訂閱這些改動(dòng)的值。都可以使用 SharedFlow 來實(shí)現(xiàn)

而關(guān)于 SharedFlow、StateFlow、LiveData的對(duì)比，個(gè)人的結(jié)論是：根據(jù)不同的場(chǎng)景 LiveData StateFlow SharedFlow 都有自己特定的使用場(chǎng)景，誰也無法真的完全平替誰。誰也不是誰的超集，都有它們各自的有點(diǎn)和缺點(diǎn)，并不能完美覆蓋所有場(chǎng)景，所以根據(jù)使用的場(chǎng)景不同按需選擇即可。

關(guān)于StateFlow 與 SharedFlow 的實(shí)戰(zhàn)，后面會(huì)總結(jié)一期。

以上就是Kotlin Flow封裝類SharedFlow StateFlow LiveData使用對(duì)比的詳細(xì)內(nèi)容，更多關(guān)于Kotlin Flow封裝類的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: