KotlinFlow背壓和線程切換竟然如此相似-創(chuàng)新互聯(lián)

前言

上篇分析了Kotlin Flow原理，大部分操作符實(shí)現(xiàn)比較簡(jiǎn)單，相較而言背壓和線程切換比較復(fù)雜，遺憾的是，縱觀網(wǎng)上大部分文章，關(guān)于Flow背壓和協(xié)程切換這塊的原理說得比較少，語焉不詳，鑒于此，本篇重點(diǎn)分析兩者的原理及使用。
通過本篇文章，你將了解到：

創(chuàng)新互聯(lián)公司長(zhǎng)期為千余家客戶提供的網(wǎng)站建設(shè)服務(wù)，團(tuán)隊(duì)從業(yè)經(jīng)驗(yàn)10年，關(guān)注不同地域、不同群體，并針對(duì)不同對(duì)象提供差異化的產(chǎn)品和服務(wù)；打造開放共贏平臺(tái)，與合作伙伴共同營(yíng)造健康的互聯(lián)網(wǎng)生態(tài)環(huán)境。為溫宿企業(yè)提供專業(yè)的網(wǎng)站建設(shè)、成都網(wǎng)站建設(shè)，溫宿網(wǎng)站改版等技術(shù)服務(wù)。擁有10多年豐富建站經(jīng)驗(yàn)和眾多成功案例,為您定制開發(fā)。
  1. 什么是背壓？
  2. 如何處理背壓？
  3. Flow buffer的原理
  4. Flow 線程切換的使用
  5. Flow 線程切換的原理
1. 什么是背壓？

先看自然界的水流：
image.png
為了充分利用水資源，人類建立了大壩，以大壩為分界點(diǎn)將水流分為上游和下游。

當(dāng)上游的流速大于下游的流速，日積月累，最終導(dǎo)致大壩溢出，此種現(xiàn)象稱為背壓的出現(xiàn)

而對(duì)于Kotlin里的Flow，也有上游（生產(chǎn)者）、下游（消費(fèi)者）的概念，如：

suspend fun testBuffer1() {var flow = flow {//生產(chǎn)者
            (1..3).forEach {println("emit $it")
                emit(it)
            }
        }

        flow.collect {//消費(fèi)者
            println("collect:$it")
        }
    }

通過collect操作符觸發(fā)了流，從生產(chǎn)者生產(chǎn)數(shù)據(jù)(flow閉包)，到消費(fèi)者接收并處理數(shù)據(jù)(collect閉包），這就完成了流從上游到下游的一次流動(dòng)過程。

2. 如何處理背壓？

模擬一個(gè)生產(chǎn)者消費(fèi)者速度不一致的場(chǎng)景：

suspend fun testBuffer3() {var flow = flow {(1..3).forEach {delay(1000)
                println("emit $it")
                emit(it)
            }
        }

        var time = measureTimeMillis {flow.collect {delay(2000)
                println("collect:$it")
            }
        }
        println("use time:${time} ms")
    }

計(jì)算流從生產(chǎn)到消費(fèi)的整個(gè)時(shí)間：
image.png
生產(chǎn)者的速度比消費(fèi)者的速度快，而它倆都是在同一個(gè)線程里順序執(zhí)行的，生產(chǎn)者必須等待消費(fèi)者消費(fèi)完畢后才會(huì)進(jìn)行下一次生產(chǎn)。
因此，整個(gè)流的耗時(shí)=生產(chǎn)者耗時(shí)(3 * 1000ms)+消費(fèi)者耗時(shí)(3 * 2000ms)=9s。

顯而易見，消費(fèi)者影響了生產(chǎn)者的速度，這種情況下該怎么優(yōu)化呢？
最簡(jiǎn)單的解決方案：

生產(chǎn)者和消費(fèi)者分別在不同的線程執(zhí)行

如：

suspend fun testBuffer4() {var flow = flow {(1..3).forEach {delay(1000)
                println("emit $it in thread:${Thread.currentThread()}")
                emit(it)
            }
        }.flowOn(Dispatchers.IO)

        var time = measureTimeMillis {flow.collect {delay(2000)
                println("collect:$it in thread:${Thread.currentThread()}")
            }
        }
        println("use time:${time} ms")
    }

添加了flowOn()函數(shù)，它的存在使得它前面的代碼在指定的線程里執(zhí)行，如flow閉包了的代碼都在IO線程執(zhí)行，也就是生產(chǎn)者在IO線程執(zhí)行。
而消費(fèi)者在當(dāng)前線程執(zhí)行，因此兩者無需相互等待，節(jié)省了總時(shí)間：
image.png

確實(shí)是減少了時(shí)間，提升了效率。但我們知道開啟線程代價(jià)還是挺大的，既然都在協(xié)程里運(yùn)行了，能否借助協(xié)程的特性：協(xié)程掛起不阻塞線程 來完成此事呢？
此時(shí)，Buffer出場(chǎng)了，先看看它是如何表演的：

suspend fun testBuffer5() {var flow = flow {(1..3).forEach {delay(1000)
                println("emit $it in thread:${Thread.currentThread()}")
                emit(it)
            }
        }.buffer(5)

        var time = measureTimeMillis {flow.collect {delay(2000)
                println("collect:$it in thread:${Thread.currentThread()}")
            }
        }
        println("use time:${time} ms")
    }

這次沒有使用flowOn，取而代之的是buffer。
運(yùn)行結(jié)果如下：
image.png
可以看出，生產(chǎn)者消費(fèi)者都是在同一線程執(zhí)行，但總耗時(shí)卻和不在同一線程運(yùn)行時(shí)相差無幾。
那么它是如何做到的呢?這就得從buffer的源碼說起。

3. Flow buffer的原理 無buffer

先看看沒有buffer時(shí)的耗時(shí)：

suspend fun testBuffer3() {var flow = flow {(1..3).forEach {delay(1000)
                println("emit $it")
                emit(it)
            }
        }

        var time = measureTimeMillis {flow.collect {delay(2000)
                println("collect:$it")
            }
        }
        println("use time:${time} ms")
    }

image.png
從collect開始，依次執(zhí)行flow閉包，通過emit調(diào)用到collect閉包，因?yàn)閒low閉包里包含了幾次emit，因此整個(gè)流程會(huì)有幾次發(fā)射。
如上圖，從步驟1到步驟8，因?yàn)槭窃谕粋€(gè)線程里，因此是串行執(zhí)行的，整個(gè)流的耗時(shí)即為生產(chǎn)者到消費(fèi)者(步驟1~步驟8）的耗時(shí)。

有buffer

在沒看源碼之前，我們先猜測(cè)一下它的流程：
image.png
每次emit都發(fā)送到buffer里，然后立刻回來繼續(xù)發(fā)送，如此一來生產(chǎn)者沒有被消費(fèi)者的速度拖累。
而消費(fèi)者會(huì)檢測(cè)Buffer里是否有數(shù)據(jù)，有則取出來。

根據(jù)之前的經(jīng)驗(yàn)我們知道：collect調(diào)用到emit最后到buffer是線性調(diào)用的，放入buffer后繼續(xù)循環(huán)emit，那么問題來了：

是誰觸發(fā)了collect閉包的調(diào)用呢？

接下來深入源碼，探究答案。

buffer源碼流程分析

創(chuàng)建Flow

public funFlow.buffer(capacity: Int = Channel.BUFFERED, onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND): Flow{var capacity = capacity//buffer容量
    var onBufferOverflow = onBufferOverflow//buffer滿之后的處理策略
    if (capacity == Channel.CONFLATED) {capacity = 0
        onBufferOverflow = BufferOverflow.DROP_OLDEST
    }
    // create a flow
    return when (this) {is FusibleFlow ->fuse(capacity = capacity, onBufferOverflow = onBufferOverflow)
        //走else 分支，構(gòu)造ChannelFlowOperatorImpl
        else ->ChannelFlowOperatorImpl(this, capacity = capacity, onBufferOverflow = onBufferOverflow)
    }
}

buffer 返回Flow實(shí)例，其間涉及幾個(gè)重要的類和函數(shù)：
image.png

調(diào)用collect
當(dāng)調(diào)用Flow.collect時(shí)：

public suspend inline funFlow.collect(crossinline action: suspend (value: T) ->Unit): Unit =
    collect(object : FlowCollector{override suspend fun emit(value: T) = action(value)
    })

構(gòu)造了匿名內(nèi)部類FlowCollector，并實(shí)現(xiàn)了emit方法，它的實(shí)現(xiàn)為collect的閉包。

調(diào)用ChannelFlowOperatorImpl.collect最終會(huì)調(diào)用ChannelFlow.collect：

override suspend fun collect(collector: FlowCollector): Unit =
        coroutineScope {collector.emitAll(produceImpl(this))
        }

    public open fun produceImpl(scope: CoroutineScope): ReceiveChannel=
        scope.produce(context, produceCapacity, onBufferOverflow, start = CoroutineStart.ATOMIC, block = collectToFun)

produceImpl 創(chuàng)建了Channel，內(nèi)部開啟了協(xié)程，返回ReceiveChannel。

再來看emitAll函數(shù)：

private suspend funFlowCollector.emitAllImpl(channel: ReceiveChannel, consume: Boolean) {ensureActive()
    var cause: Throwable? = null
    try {while (true) {//掛起等待Channel數(shù)據(jù)
            val result = run {channel.receiveCatching() }
            if (result.isClosed) {//Channel關(guān)閉后才會(huì)退出循環(huán)
                result.exceptionOrNull()?.let {throw it }
                break // returns normally when result.closeCause == null
            }
            //發(fā)送數(shù)據(jù)
            emit(result.getOrThrow())
        }
    } catch (e: Throwable) {cause = e
        throw e
    } finally {if (consume) channel.cancelConsumed(cause)
    }
}

Channel此時(shí)并沒有數(shù)據(jù)，因此協(xié)程會(huì)掛起等待。

Channel發(fā)送
Channel什么時(shí)候有數(shù)據(jù)呢？當(dāng)然是在調(diào)用了Channel.send()函數(shù)后。
前面提到過collect之后開啟了協(xié)程：

public open fun produceImpl(scope: CoroutineScope): ReceiveChannel=
        scope.produce(context, produceCapacity, onBufferOverflow, start = CoroutineStart.ATOMIC, block = collectToFun)

  internal val collectToFun: suspend (ProducerScope) ->Unit
        get() = {collectTo(it) }

  protected override suspend fun collectTo(scope: ProducerScope) =
        flowCollect(SendingCollector(scope))

此時(shí)傳入的參數(shù)為：collectToFun，最后構(gòu)造了：

public class SendingCollector(
    private val channel: SendChannel) : FlowCollector{override suspend fun emit(value: T): Unit = channel.send(value)
}

當(dāng)協(xié)程得到執(zhí)行時(shí)，會(huì)調(diào)用collectToFun–>collectTo(it)–>flowCollect(SendingCollector(scope))，最終調(diào)用到：

#ChannelFlowOperatorImpl
    override suspend fun flowCollect(collector: FlowCollector) =
        flow.collect(collector)

而該flow為最開始的flow，collector為SendingCollector。
flow.collect后會(huì)調(diào)用到flow的閉包，進(jìn)而調(diào)用到emit函數(shù)：

private fun emit(uCont: Continuation, value: T): Any? {val currentContext = uCont.context
        currentContext.ensureActive()
        //...
        completion = uCont
        return emitFun(collector as FlowCollector, value, this as Continuation)
    }

emitFun本質(zhì)上會(huì)調(diào)用collector里的emit函數(shù)，而此時(shí)的collector即為SendingCollector，最后調(diào)用channel.send(value)

如此一來，Channel就將數(shù)據(jù)發(fā)送出去了，此時(shí)channel.receiveCatching()被喚醒，接下來執(zhí)行emit(result.getOrThrow())，這函數(shù)最后會(huì)流轉(zhuǎn)到最初始的collect的閉包里。
上面的分析即為生產(chǎn)者到消費(fèi)者的流轉(zhuǎn)過程，單看源碼可能比較亂，看圖解惑：

image.png
紅色部分和綠色部分分別為不同的協(xié)程，它倆的關(guān)聯(lián)點(diǎn)即是藍(lán)色部分。

Flow buffer的本質(zhì)上是利用了Channel進(jìn)行數(shù)據(jù)的發(fā)送和接收

buffer為啥能提升效率

前面分析過無buffer時(shí)生產(chǎn)者消費(fèi)者的流程圖，作為對(duì)比，我們也將加入buffer后生產(chǎn)者消費(fèi)者的流程圖。
image.png
還是以相同的demo，闡述其流程：

  1. 生產(chǎn)者掛起1s，當(dāng)1s結(jié)束后調(diào)用emit發(fā)射數(shù)據(jù)，此時(shí)數(shù)據(jù)放入buffer里，生產(chǎn)者調(diào)用delay繼續(xù)掛起
  2. 此時(shí)消費(fèi)者被喚醒，然后掛起 2s等待
  3. 第2s到來之時(shí)，生產(chǎn)者調(diào)用emit發(fā)送數(shù)據(jù)到buffer里，繼續(xù)掛起
  4. 第2s到來之時(shí)，消費(fèi)者結(jié)束掛起，消費(fèi)數(shù)據(jù)，然后繼續(xù)掛起2s
  5. 第3s到來之時(shí)，生產(chǎn)者繼續(xù)生產(chǎn)數(shù)據(jù)，而后生產(chǎn)者退出生產(chǎn)
  6. 第5s到來之時(shí)，消費(fèi)者掛起結(jié)束，消費(fèi)數(shù)據(jù)，然后繼續(xù)掛起2s
  7. 第7s到來之時(shí)，消費(fèi)者掛起結(jié)束，消費(fèi)結(jié)束，此時(shí)因?yàn)閏hannel里已經(jīng)沒有數(shù)據(jù)了，退出循環(huán)，最終消費(fèi)者退出

由此可見，總共花費(fèi)了7s。
image.png
ps：協(xié)程調(diào)度時(shí)機(jī)不同，打印順序可能略有差異，但總體耗時(shí)不變。

至此，我們找到了buffer能夠提高效率的原因：

生產(chǎn)者、消費(fèi)者運(yùn)行在不同的協(xié)程，掛起操作不阻塞對(duì)方

拋出一個(gè)比較有意思的問題：以下代碼加buffer之后效率會(huì)有提升嗎？

suspend fun testBuffer6() {var flow = flow {(1..3).forEach {println("emit $it")
                emit(it)
            }
        }
        var time = measureTimeMillis {flow.collect {delay(2000)
                println("collect:$it")
            }
        }
        println("use time:${time} ms")
    }

在未實(shí)驗(yàn)之前，如果你已經(jīng)有答案，恭喜你已經(jīng)弄懂了buffer的本質(zhì)。

4. Flow 線程切換的使用
suspend fun testBuffer4() {var flow = flow {(1..3).forEach {delay(1000)
                println("emit $it in thread:${Thread.currentThread()}")
                emit(it)
            }
        }.flowOn(Dispatchers.IO)

        var time = measureTimeMillis {flow.collect {delay(2000)
                println("collect:$it in thread:${Thread.currentThread()}")
            }
        }
        println("use time:${time} ms")
    }

flowOn(Dispatchers.IO)表示其之前的操作符(函數(shù))都在IO線程執(zhí)行，如這里的意思是flow閉包里的代碼在IO線程執(zhí)行。
而其之后的操作符(函數(shù))在當(dāng)前的線程執(zhí)行。
通常用在子線程里獲取網(wǎng)絡(luò)數(shù)據(jù)(flow閉包)，然后再collect閉包里(主線程)更新UI。

5. Flow 線程切換的原理
public funFlow.flowOn(context: CoroutineContext): Flow{checkFlowContext(context)
    return when {context == EmptyCoroutineContext ->this
        this is FusibleFlow ->fuse(context = context)
        else ->ChannelFlowOperatorImpl(this, context = context)
    }
}

看到這你可能已經(jīng)有答案了：這不就和buffer一樣的方式嗎？
但仔細(xì)看，此處多了個(gè)上下文：CoroutineContext。
CoroutineContext的作用就是用來決定協(xié)程運(yùn)行在哪個(gè)線程。

前面分析的buffer時(shí)，我們的協(xié)程的作用域是runBlocking，即使生產(chǎn)者、消費(fèi)者在不同的協(xié)程，但是它們始終在同一個(gè)線程里執(zhí)行。
而使用了flowOn指定線程，此時(shí)生產(chǎn)者、消費(fèi)者在不同的線程運(yùn)行協(xié)程。
因此，只要弄懂了buffer原理，flowOn原理自然而然就懂了。
image.png

以上為Flow背壓和線程切換的全部?jī)?nèi)容，下篇將分析Flow的熱流。
本文基于Kotlin 1.5.3，文中完整Demo請(qǐng)點(diǎn)擊

您若喜歡，請(qǐng)點(diǎn)贊、關(guān)注、收藏，您的鼓勵(lì)是我前進(jìn)的動(dòng)力 持續(xù)更新中，和我一起步步為營(yíng)系統(tǒng)、深入學(xué)習(xí)Android/Kotlin

1、Android各種Context的前世今生
2、Android DecorView 必知必會(huì)
3、Window/WindowManager 不可不知之事
4、View Measure/Layout/Draw 真明白了
5、Android事件分發(fā)全套服務(wù)
6、Android invalidate/postInvalidate/requestLayout 徹底厘清
7、Android Window 如何確定大小/onMeasure()多次執(zhí)行原因
8、Android事件驅(qū)動(dòng)Handler-Message-Looper解析
9、Android 鍵盤一招搞定
10、Android 各種坐標(biāo)徹底明了
11、Android Activity/Window/View 的background
12、Android Activity創(chuàng)建到View的顯示過
13、Android IPC 系列
14、Android 存儲(chǔ)系列
15、Java 并發(fā)系列不再疑惑
16、Java 線程池系列
17、Android Jetpack 前置基礎(chǔ)系列
18、Android Jetpack 易學(xué)易懂系列
19、Kotlin 輕松入門系列
20、Kotlin 協(xié)程系列全面解讀

你是否還在尋找穩(wěn)定的海外服務(wù)器提供商？創(chuàng)新互聯(lián)www.cdcxhl.cn海外機(jī)房具備T級(jí)流量清洗系統(tǒng)配攻擊溯源，準(zhǔn)確流量調(diào)度確保服務(wù)器高可用性，企業(yè)級(jí)服務(wù)器適合批量采購(gòu)，新人活動(dòng)首月15元起，快前往官網(wǎng)查看詳情吧

名稱欄目：KotlinFlow背壓和線程切換竟然如此相似-創(chuàng)新互聯(lián)
當(dāng)前路徑：http://www.muchs.cn/article42/ceooec.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供手機(jī)網(wǎng)站建設(shè)、網(wǎng)頁(yè)設(shè)計(jì)公司、小程序開發(fā)、網(wǎng)站內(nèi)鏈、企業(yè)建站、外貿(mào)建站

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來源： 創(chuàng)新互聯(lián)

微信小程序開發(fā)