KotlinFlow背壓和線程切換竟然如此相似-創(chuàng)新互聯(lián)
上篇分析了Kotlin Flow原理，大部分操作符實現(xiàn)比較簡單，相較而言背壓和線程切換比較復雜，遺憾的是，縱觀網(wǎng)上大部分文章，關于Flow背壓和協(xié)程切換這塊的原理說得比較少，語焉不詳，鑒于此，本篇重點分析兩者的原理及使用。
通過本篇文章，你將了解到：
1. 什么是背壓？
- 什么是背壓？
- 如何處理背壓？
- Flow buffer的原理
- Flow 線程切換的使用
- Flow 線程切換的原理
先看自然界的水流：
為了充分利用水資源，人類建立了大壩，以大壩為分界點將水流分為上游和下游。
當上游的流速大于下游的流速，日積月累，最終導致大壩溢出，此種現(xiàn)象稱為背壓的出現(xiàn)
而對于Kotlin里的Flow，也有上游（生產(chǎn)者）、下游（消費者）的概念，如：
suspend fun testBuffer1() {var flow = flow {//生產(chǎn)者
(1..3).forEach {println("emit $it")
emit(it)
}
}
flow.collect {//消費者
println("collect:$it")
}
}通過collect操作符觸發(fā)了流，從生產(chǎn)者生產(chǎn)數(shù)據(jù)(flow閉包)，到消費者接收并處理數(shù)據(jù)(collect閉包），這就完成了流從上游到下游的一次流動過程。
2. 如何處理背壓？模擬一個生產(chǎn)者消費者速度不一致的場景：
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")
}計算流從生產(chǎn)到消費的整個時間：
生產(chǎn)者的速度比消費者的速度快，而它倆都是在同一個線程里順序執(zhí)行的，生產(chǎn)者必須等待消費者消費完畢后才會進行下一次生產(chǎn)。
因此，整個流的耗時=生產(chǎn)者耗時(3 * 1000ms)+消費者耗時(3 * 2000ms)=9s。
顯而易見，消費者影響了生產(chǎn)者的速度，這種情況下該怎么優(yōu)化呢？
最簡單的解決方案：
生產(chǎn)者和消費者分別在不同的線程執(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í)行。
而消費者在當前線程執(zhí)行，因此兩者無需相互等待，節(jié)省了總時間：
確實是減少了時間，提升了效率。但我們知道開啟線程代價還是挺大的，既然都在協(xié)程里運行了，能否借助協(xié)程的特性：協(xié)程掛起不阻塞線程 來完成此事呢？
此時，Buffer出場了，先看看它是如何表演的：
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。
運行結果如下：
可以看出，生產(chǎn)者消費者都是在同一線程執(zhí)行，但總耗時卻和不在同一線程運行時相差無幾。
那么它是如何做到的呢?這就得從buffer的源碼說起。
先看看沒有buffer時的耗時：
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")
}
從collect開始，依次執(zhí)行flow閉包，通過emit調用到collect閉包，因為flow閉包里包含了幾次emit，因此整個流程會有幾次發(fā)射。
如上圖，從步驟1到步驟8，因為是在同一個線程里，因此是串行執(zhí)行的，整個流的耗時即為生產(chǎn)者到消費者(步驟1~步驟8）的耗時。
在沒看源碼之前，我們先猜測一下它的流程：
每次emit都發(fā)送到buffer里，然后立刻回來繼續(xù)發(fā)送，如此一來生產(chǎn)者沒有被消費者的速度拖累。
而消費者會檢測Buffer里是否有數(shù)據(jù)，有則取出來。
根據(jù)之前的經(jīng)驗我們知道：collect調用到emit最后到buffer是線性調用的，放入buffer后繼續(xù)循環(huán)emit，那么問題來了：
是誰觸發(fā)了collect閉包的調用呢？
接下來深入源碼，探究答案。
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 分支，構造ChannelFlowOperatorImpl
else ->ChannelFlowOperatorImpl(this, capacity = capacity, onBufferOverflow = onBufferOverflow)
}
} buffer 返回Flow實例，其間涉及幾個重要的類和函數(shù)：
調用collect
當調用Flow.collect時：
public suspend inline funFlow.collect(crossinline action: suspend (value: T) ->Unit): Unit =
collect(object : FlowCollector{override suspend fun emit(value: T) = action(value)
}) 構造了匿名內部類FlowCollector，并實現(xiàn)了emit方法，它的實現(xiàn)為collect的閉包。
調用ChannelFlowOperatorImpl.collect最終會調用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，內部開啟了協(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關閉后才會退出循環(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ù)據(jù)，因此協(xié)程會掛起等待。
Channel發(fā)送
Channel什么時候有數(shù)據(jù)呢？當然是在調用了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ù)為：collectToFun，最后構造了：
public class SendingCollector(
private val channel: SendChannel) : FlowCollector{override suspend fun emit(value: T): Unit = channel.send(value)
} 當協(xié)程得到執(zhí)行時，會調用collectToFun–>collectTo(it)–>flowCollect(SendingCollector(scope))，最終調用到：
#ChannelFlowOperatorImpl
override suspend fun flowCollect(collector: FlowCollector) =
flow.collect(collector) 而該flow為最開始的flow，collector為SendingCollector。
flow.collect后會調用到flow的閉包，進而調用到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本質上會調用collector里的emit函數(shù)，而此時的collector即為SendingCollector，最后調用channel.send(value)
如此一來，Channel就將數(shù)據(jù)發(fā)送出去了，此時channel.receiveCatching()被喚醒，接下來執(zhí)行emit(result.getOrThrow())，這函數(shù)最后會流轉到最初始的collect的閉包里。
上面的分析即為生產(chǎn)者到消費者的流轉過程，單看源碼可能比較亂，看圖解惑：
紅色部分和綠色部分分別為不同的協(xié)程，它倆的關聯(lián)點即是藍色部分。
buffer為啥能提升效率Flow buffer的本質上是利用了Channel進行數(shù)據(jù)的發(fā)送和接收
前面分析過無buffer時生產(chǎn)者消費者的流程圖，作為對比，我們也將加入buffer后生產(chǎn)者消費者的流程圖。
還是以相同的demo，闡述其流程：
- 生產(chǎn)者掛起1s，當1s結束后調用emit發(fā)射數(shù)據(jù)，此時數(shù)據(jù)放入buffer里，生產(chǎn)者調用delay繼續(xù)掛起
- 此時消費者被喚醒，然后掛起 2s等待
- 第2s到來之時，生產(chǎn)者調用emit發(fā)送數(shù)據(jù)到buffer里，繼續(xù)掛起
- 第2s到來之時，消費者結束掛起，消費數(shù)據(jù)，然后繼續(xù)掛起2s
- 第3s到來之時，生產(chǎn)者繼續(xù)生產(chǎn)數(shù)據(jù)，而后生產(chǎn)者退出生產(chǎn)
- 第5s到來之時，消費者掛起結束，消費數(shù)據(jù)，然后繼續(xù)掛起2s
- 第7s到來之時，消費者掛起結束，消費結束，此時因為channel里已經(jīng)沒有數(shù)據(jù)了，退出循環(huán)，最終消費者退出
由此可見，總共花費了7s。
ps：協(xié)程調度時機不同，打印順序可能略有差異，但總體耗時不變。
至此，我們找到了buffer能夠提高效率的原因：
生產(chǎn)者、消費者運行在不同的協(xié)程，掛起操作不阻塞對方
拋出一個比較有意思的問題：以下代碼加buffer之后效率會有提升嗎？
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")
}在未實驗之前，如果你已經(jīng)有答案，恭喜你已經(jīng)弄懂了buffer的本質。
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ù))在當前的線程執(zhí)行。
通常用在子線程里獲取網(wǎng)絡數(shù)據(jù)(flow閉包)，然后再collect閉包里(主線程)更新UI。
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一樣的方式嗎？
但仔細看，此處多了個上下文：CoroutineContext。
CoroutineContext的作用就是用來決定協(xié)程運行在哪個線程。
前面分析的buffer時，我們的協(xié)程的作用域是runBlocking，即使生產(chǎn)者、消費者在不同的協(xié)程，但是它們始終在同一個線程里執(zhí)行。
而使用了flowOn指定線程，此時生產(chǎn)者、消費者在不同的線程運行協(xié)程。
因此，只要弄懂了buffer原理，flowOn原理自然而然就懂了。
以上為Flow背壓和線程切換的全部內容，下篇將分析Flow的熱流。
本文基于Kotlin 1.5.3，文中完整Demo請點擊
1、Android各種Context的前世今生
2、Android DecorView 必知必會
3、Window/WindowManager 不可不知之事
4、View Measure/Layout/Draw 真明白了
5、Android事件分發(fā)全套服務
6、Android invalidate/postInvalidate/requestLayout 徹底厘清
7、Android Window 如何確定大小/onMeasure()多次執(zhí)行原因
8、Android事件驅動Handler-Message-Looper解析
9、Android 鍵盤一招搞定
10、Android 各種坐標徹底明了
11、Android Activity/Window/View 的background
12、Android Activity創(chuàng)建到View的顯示過
13、Android IPC 系列
14、Android 存儲系列
15、Java 并發(fā)系列不再疑惑
16、Java 線程池系列
17、Android Jetpack 前置基礎系列
18、Android Jetpack 易學易懂系列
19、Kotlin 輕松入門系列
20、Kotlin 協(xié)程系列全面解讀
你是否還在尋找穩(wěn)定的海外服務器提供商？創(chuàng)新互聯(lián)www.cdcxhl.cn海外機房具備T級流量清洗系統(tǒng)配攻擊溯源，準確流量調度確保服務器高可用性，企業(yè)級服務器適合批量采購，新人活動首月15元起，快前往官網(wǎng)查看詳情吧
分享文章：KotlinFlow背壓和線程切換竟然如此相似-創(chuàng)新互聯(lián)
文章來源：http://m.jiaotiyi.com/article/ceooec.html