欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

圖解 Kotlin SharedFlow 緩存系統(tǒng)及示例詳解

 更新時(shí)間：2022年10月20日 11:04:02   作者：fundroid  
這篇文章主要為大家介紹了圖解 Kotlin SharedFlow 緩存系統(tǒng)及示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

前言

Kotlin 為我們提供了兩種創(chuàng)建“熱流”的工具：StateFlow 和 SharedFlow。StateFlow 經(jīng)常被用來(lái)替代 LiveData 充當(dāng)架構(gòu)組件使用，所以大家相對(duì)熟悉。其實(shí) StateFlow 只是 SharedFlow 的一種特化形式，SharedFlow 的功能更強(qiáng)大、使用場(chǎng)景更多，這得益于其自帶的緩存系統(tǒng)，本文用圖解的方式，帶大家更形象地理解 SharedFlow 的緩存系統(tǒng)。

創(chuàng)建 SharedFlow 需要使用到 MutableSharedFlow() 方法，我們通過(guò)方法的三個(gè)參數(shù)配置緩存：

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

接下來(lái)，我們通過(guò)時(shí)序圖的形式介紹這三個(gè)關(guān)鍵參數(shù)對(duì)緩存的影響。正文之前讓我們先統(tǒng)一一下用語(yǔ)：

  • Emitter：Flow 數(shù)據(jù)的生產(chǎn)者，從上游發(fā)射數(shù)據(jù)
  • Subcriber：Flow 數(shù)據(jù)的消費(fèi)者，在下游接收數(shù)據(jù)

replay

當(dāng) Subscriber 訂閱 SharedFlow 時(shí)，有機(jī)會(huì)接收到之前已發(fā)送過(guò)的數(shù)據(jù)，replay 指定了可以收到 subscribe 之前數(shù)據(jù)的數(shù)量。replay 不能為負(fù)數(shù)，默認(rèn)值為 0 表示 Subscriber 只能接收到 subscribe 之后 emit 的數(shù)據(jù)：

上圖展示的是 replay = 0 的情況，Subscriber 無(wú)法收到 subscribe 之前 emit 的 ?，只能接收到 ? 和 ?。

當(dāng) replay = n （ n > 0）時(shí)，SharedFlow 會(huì)啟用緩存，此時(shí) BufferSize 為 n，意味著可以緩存發(fā)射過(guò)的最近 n 個(gè)數(shù)據(jù)，并發(fā)送給新增的 Subscriber。

上圖以 n = 1 為例 ：

  • Emitter 發(fā)送 ? ，并被 Buffer 緩存
  • Subscriber 訂閱 SharedFlow 后，接收到緩存的 ?
  • Emitter 相繼發(fā)送 ? ? ，Buffer 緩存的數(shù)據(jù)相繼依次被更新

在生產(chǎn)者消費(fèi)者模型中，有時(shí)消費(fèi)的速度趕不及生產(chǎn)，此時(shí)要加以控制，要么停止生產(chǎn)，要么丟棄數(shù)據(jù)。SharedFlow 也同樣如此。有時(shí) Subscriber 的處理速度較慢，Buffer 緩存的數(shù)據(jù)得不到及時(shí)處理，當(dāng) Buffer 為空時(shí)，emit 默認(rèn)將會(huì)被掛起 （ onBufferOverflow = SUSPEND）

上面的圖展示了 replay = 1 時(shí) emit 發(fā)生 suspend 場(chǎng)景：

  • Emitter 發(fā)送 ? 并被緩存
  • Subscriber 訂閱 SharedFlow ，接收 replay 的 ? 開(kāi)始處理
  • Emitter 發(fā)送 ? ，緩存數(shù)據(jù)更新為 ? ，由于 Subscriber 對(duì) ? 的處理尚未結(jié)束，? 在緩存中沒(méi)有及時(shí)被消費(fèi)
  • Emitter 發(fā)送 ?，由于緩存的 ? 尚未被 Subscriber 消費(fèi)，emit 發(fā)生掛起
  • Subscriber 開(kāi)始消費(fèi) ? ，Buffer 緩存 ? ， Emitter 可以繼續(xù) emit 新數(shù)據(jù)

注意 SharedFlow 作為一個(gè)多播可以有多個(gè) Subscriber，所以上面例子中，? 被消費(fèi)的時(shí)間點(diǎn)，取決于最后一個(gè)開(kāi)始處理的 Subscriber。

extraBufferCapacity

extraBufferCapacity 中的 extra 表示 replay-cache 之外為 Buffer 還可以額外追加的緩存。

若 replay = n， extraBufferCapacity = m，則 BufferSize = m + n。

extraBufferCapacity 默認(rèn)為 0，設(shè)置 extraBufferCapacity 有助于提升 Emitter 的吞吐量

在上圖的基礎(chǔ)之上，我們?cè)僭O(shè)置 extraBufferCapacity = 1，效果如下圖：

上圖中 BufferSize = 1 + 1 = 2 ：

  • Emitter 發(fā)送 ? 并得到 Subscriber1 的處理 ，? 作為 replay 的一個(gè)數(shù)據(jù)被緩存，
  • Emitter 發(fā)送 ?，Buffer 中 replay-cache 的數(shù)據(jù)更新為 ?
  • Emitter 發(fā)送 ?，Buffer 在存儲(chǔ)了 replay 數(shù)據(jù) ? 之上，作為 extra 又存儲(chǔ)了 ?
  • Emitter 發(fā)送 ?，此時(shí) Buffer 已沒(méi)有空余位置，emit 掛起
  • Subscriber2 訂閱 SharedFlow。雖然此時(shí) Buffer 中存有 ? ? 兩個(gè)數(shù)據(jù)，但是由于 replay = 1，所以 Subscriber2 只能收到最近的一個(gè)數(shù)據(jù) ?
  • Subscriber1 處理完 ? 后，依次處理 Buffer 中的下一個(gè)數(shù)據(jù)，開(kāi)始消費(fèi) ?
  • 對(duì)于 SharedFlow 來(lái)說(shuō)，已經(jīng)不存在沒(méi)有消費(fèi) ? 的 Subscriber，? 移除緩存，? 的 emit 繼續(xù)，并進(jìn)入緩存，此時(shí) Buffer 又有兩個(gè)數(shù)據(jù) ? ? ，
  • Subscriber1 處理完 ? ，開(kāi)始消費(fèi) ?
  • 不存在沒(méi)有消費(fèi) ? 的 Subscriber， ? 移除緩存。

onBufferOverflow

前面的例子中，當(dāng) Buffer 被填滿時(shí)，emit 會(huì)被掛起，這都是建立在 onBufferOverflow 為 SUSPEND 的前提下的。onBufferOverflow 用來(lái)指定緩存移除時(shí)的策略，除了默認(rèn)的 SUSPEND，還有兩個(gè)數(shù)據(jù)丟棄策略：

  • DROP_LATEST：丟棄最新的數(shù)據(jù)
  • DROP_OLDEST：丟棄最老的數(shù)據(jù)

需要特別注意的是，當(dāng) BufferSize = 0 時(shí)，extraBufferCapacity 只支持 SUSPEND，其他丟棄策略是無(wú)效的。這很好理解，因?yàn)?Buffer 中沒(méi)有數(shù)據(jù)，所以丟棄無(wú)從下手，所以啟動(dòng)丟棄策略的前提是 Buffer 至少有一個(gè)緩沖區(qū)，且數(shù)據(jù)被填滿

上圖展示 DROP_LATEST 的效果。假設(shè) replay = 2，extra = 0

  • Emitter 發(fā)送 ? 時(shí)，由于 ? 已經(jīng)被消費(fèi)，所以 Buffer 數(shù)據(jù)從 ?? 變?yōu)???
  • Emitter 發(fā)送 ? 時(shí)，由于 ? 還未被消費(fèi)，Buffer 處于填滿狀態(tài)， ? 直接被丟棄
  • Emitter 發(fā)送 ? 時(shí)，由于 ? 已經(jīng)被費(fèi)，可以移除緩存，Buffer 數(shù)據(jù)變?yōu)???

上圖展示了 DROP_OLDEST 的效果，與 DROP_LATEST 比較后非常明顯，緩存中永遠(yuǎn)會(huì)儲(chǔ)存最新的兩個(gè)數(shù)據(jù)，但是較老的數(shù)據(jù)不管有沒(méi)有被消費(fèi)，都可能會(huì)從 Buffer 移除，所以 Subscriber 可以消費(fèi)當(dāng)前最新的數(shù)據(jù)，但是有可能漏掉中間的數(shù)據(jù)，比如圖中漏掉了 ?

注意：當(dāng) extraBufferCapacity 設(shè)為 SUSPEND 可以保證 Subscriber 一個(gè)不漏的消費(fèi)掉所有數(shù)據(jù)，但是會(huì)影響 Emitter 的速度；當(dāng)設(shè)置為 DROP_XXX 時(shí)，可以保證 emit 調(diào)用后立即返回，但是 Subscriber 可能會(huì)漏掉部分?jǐn)?shù)據(jù)。

如果我們不想讓 emit 發(fā)生掛起，除了設(shè)置 DROP_XXX 之外，還有一個(gè)方法就是調(diào)用 tryEmit，這是一個(gè)非 suspend 版本的 emit

abstract suspend override fun emit(value: T)
abstract fun tryEmit(value: T): Boolean

tryEmit 返回一個(gè) boolean 值，你可以這樣判斷返回值，當(dāng)使用 emit 會(huì)掛起時(shí)，使用 tryEmit 會(huì)返回 false，其余情況都是 true。這意味著 tryEmit 返回 false 的前提是 extraBufferCapacity 必須設(shè)為 SUSPEND，且 Buffer 中空余位置為 0 。此時(shí)使用 tryEmit 的效果等同于 DROP_LATEST。

SharedFlow Buffer

前面介紹的 MutableSharedFlow 的三個(gè)參數(shù)，其本質(zhì)都是圍繞 SharedFlow 的 Buffer 進(jìn)行工作的。那么這個(gè) Buffer 具體結(jié)構(gòu)是怎樣的呢？

上面這個(gè)圖是 SharedFlow 源碼中關(guān)于 Buffer 的注釋，這個(gè)圖形象地告訴了我們 Buffer 是一個(gè)線性數(shù)據(jù)結(jié)構(gòu)（就是一個(gè)普通的數(shù)組 Array<Any?>），但是這個(gè)圖不能直觀反應(yīng) Buffer 運(yùn)行機(jī)制。下面通過(guò)一個(gè)例子，看一下 Buffer 在運(yùn)行時(shí)的具體更新過(guò)程：

val sharedFlow = MutableSharedFlow<Int>(
    replay = 2, 
    extraBufferCapacity = 2,
    onBufferOverflow = BufferOverflow.SUSPEND
)
var emitValue = 1
fun main() {
    runBlocking {
        launch {
            sharedFlow.onEach {
                delay(200) // simulate the consume of data
            }.collect()
        }
        repeat(12) {
            sharedFlow.emit(emitValue)
            emitValue++
            delay(50)
        }
    }
}

上面的代碼很簡(jiǎn)單，SharedFlow 的 BufferSize = 2+2 = 4，Emitter 生產(chǎn)的速度大于 Subscriber 消費(fèi)的速度，所以過(guò)程中會(huì)出現(xiàn) Buffer 的填充和更新，下面依舊用圖的方式展示 Buffer 的變化

先看一下代碼對(duì)應(yīng)的時(shí)序圖:

有前面的介紹，相信這個(gè)時(shí)序圖很容易理解，這里就不再贅述了，下面重點(diǎn)圖解一下 Buffer 的內(nèi)存變化。SharedFlow 的 Buffer 本質(zhì)上是一個(gè)基于 Array 實(shí)現(xiàn)的 queue，通過(guò)指針移動(dòng)從往隊(duì)列增刪元素，避免了元素在實(shí)際數(shù)組中的移動(dòng)。這里關(guān)鍵的指針有三個(gè)：

  • head：隊(duì)列的 head 指向 Buffer 的第一個(gè)有效數(shù)據(jù)，這是時(shí)間上最早進(jìn)入緩存的數(shù)據(jù)，在數(shù)據(jù)被所有的 Subscriber 消費(fèi)之前不會(huì)移除緩存。因此 head 也代表了最慢的 Subscriber 的處理進(jìn)度
  • replay：Buffer 為 replay-cache 預(yù)留空間的其實(shí)位置，當(dāng)有新的 Subscriber 訂閱發(fā)生時(shí)，從此位置開(kāi)始處理數(shù)據(jù)。
  • end：新數(shù)據(jù)進(jìn)入緩存時(shí)的位置，end 這也代表了最快的 Subscriber 的處理進(jìn)度。

如果 bufferSize 表示當(dāng)前 Buffer 中存儲(chǔ)數(shù)據(jù)的個(gè)數(shù)，則我們可知三指針 index 符合如下關(guān)系：

  • replay <= head + bufferSize
  • end = head + bufferSize

了解了三指針的含義后，我們?cè)賮?lái)看上圖中的 Buffer 是如何工作的：

最后，總結(jié)一下 Buffer 的特點(diǎn)：

  • 基于數(shù)組實(shí)現(xiàn)，當(dāng)數(shù)組空間不夠時(shí)進(jìn)行 2n 的擴(kuò)容
  • 元素進(jìn)入數(shù)組后的位置保持不變，通過(guò)移動(dòng)指針，決定數(shù)據(jù)的消費(fèi)起點(diǎn)
  • 指針移動(dòng)到數(shù)組尾部后，會(huì)重新指向頭部，數(shù)組空間可循環(huán)使用

以上就是圖解 Kotlin SharedFlow 緩存系統(tǒng)及示例詳解的詳細(xì)內(nèi)容，更多關(guān)于Kotlin SharedFlow 緩存的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

相關(guān)文章

最新評(píng)論