KotlinSelect協(xié)程多路復(fù)用的實現(xiàn)詳解

第KotlinSelect協(xié)程多路復(fù)用的實現(xiàn)詳解目錄前言1.Select的引入多路數(shù)據(jù)的選擇串行執(zhí)行協(xié)程并行執(zhí)行同時監(jiān)聽多路結(jié)果Select閃亮登場2.Select的使用3.Invoke函數(shù)的妙用4.Select的原理5.Select注意事項

前言

協(xié)程通信三劍客：Channel、Select、Flow，本篇將會重點分析Select的使用及原理。

通過本篇文章，你將了解到：

Select的引入Select的使用Invoke函數(shù)的妙用Select的原理Select注意事項

1.Select的引入

多路數(shù)據(jù)的選擇

串行執(zhí)行

如今的二維碼識別應(yīng)用場景越來越廣了，早期應(yīng)用比較廣泛的識別SDK如zxing、zbar，它們各有各的特點，也存在識別不出來的情況，為了將兩者優(yōu)勢結(jié)合起來，我們想到的方法是同一份二維碼圖片分別給兩者進行識別。

如下：

//從zxing獲取二維碼信息

suspendfungetQrcodeInfoFromZxing(bitmap:Bitmap):String{

//模擬耗時

delay(2000)

return"I'mfish"

//從zbar獲取二維碼信息

suspendfungetQrcodeInfoFromZbar(bitmap:Bitmap):String{

delay(1000)

return"I'mfish"

funtestSelect(){

runBlocking{

varbitmap=null

varstarTime=System.currentTimeMillis()

varqrcoe1=getQrcodeInfoFromZxing(bitmap)

varqrcode2=getQrcodeInfoFromZbar(bitmap)

println("qrcode1=$qrcoe1qrcode2=$qrcode2useTime:${System.currentTimeMillis()-starTime}ms")

}

查看打印，最后花費的時間：

qrcode1=Imfishqrcode2=ImfishuseTime:3013ms

當(dāng)然這是串行的方式效率比較低，我們想到了用協(xié)程來優(yōu)化它。

協(xié)程并行執(zhí)行

如下：

funtestSelect1(){

varbitmap=null;

varstarTime=System.currentTimeMillis()

vardeferredZxing=GlobalScope.async{

getQrcodeInfoFromZxing(bitmap)

vardeferredZbar=GlobalScope.async{

getQrcodeInfoFromZbar(bitmap)

runBlocking{

//掛起等待識別結(jié)果

varqrcoe1=deferredZxing.await()

//掛起等待識別結(jié)果

varqrcode2=deferredZbar.await()

println("qrcode1=$qrcoe1qrcode2=$qrcode2useTime:${System.currentTimeMillis()-starTime}ms")

}

查看打印，最后花費的時間：

qrcode1=Imfishqrcode2=ImfishuseTime:2084ms

可以看出，花費時間明顯變少了。

與上個Demo相比，雖然識別過程是放在協(xié)程里并行執(zhí)行的，但是在等待識別結(jié)果卻是串行的。我們引入兩個識別庫的初衷是哪個識別快就用哪個的結(jié)果，為了達成這個目的，傳統(tǒng)的方式是：

同時監(jiān)聽并記錄識別結(jié)果的返回。

同時監(jiān)聽多路結(jié)果

如下：

funtestSelect2(){

varbitmap=null;

varstarTime=System.currentTimeMillis()

vardeferredZxing=GlobalScope.async{

getQrcodeInfoFromZxing(bitmap)

vardeferredZbar=GlobalScope.async{

getQrcodeInfoFromZbar(bitmap)

varisEnd=false

varresult:String=null

GlobalScope.launch{

if(!isEnd){

//沒有結(jié)束，則繼續(xù)識別

varresultTmp=deferredZxing.await()

if(!isEnd){

//識別沒有結(jié)束，說明自己是第一個返回結(jié)果的

result=resultTmp

println("zxingrecognizeokuseTime:${System.currentTimeMillis()-starTime}ms")

//標(biāo)記識別結(jié)束

isEnd=true

GlobalScope.launch{

if(!isEnd){

varresultTmp=deferredZbar.await()

if(!isEnd){

//識別沒有結(jié)束，說明自己是第一個返回結(jié)果的

result=resultTmp

println("zbarrecognizeokuseTime:${System.currentTimeMillis()-starTime}ms")

isEnd=true

//檢測是否有結(jié)果返回

runBlocking{

while(!isEnd){

delay(1)

println("recognizeresult:$result")

}

通過檢測isEnd標(biāo)記來判斷是否有某個模塊返回結(jié)果。

結(jié)果如下：

zbarrecognizeokuseTime:1070ms

recognizeresult:Imfish

由于模擬設(shè)定的zbar解析速度快，因此每次都是采納的是zbar的結(jié)果，所花費的時間大幅減少了，該結(jié)果符合預(yù)期。

Select閃亮登場

雖說上個Demo結(jié)果符合預(yù)期，但是多了很多額外的代碼、多引入了其它協(xié)程，并且需要子模塊對標(biāo)記進行賦值（對isEnd進行賦值），沒有達到解耦的目的。我們希望子模塊的任務(wù)是單一且閉環(huán)的，如果能在一個函數(shù)里統(tǒng)一檢測結(jié)果的返回就好了。

Select就是為了解決多路數(shù)據(jù)的選擇而生的。

來看看它是怎么解決該問題的：

funtestSelect3(){

varbitmap=null;

varstarTime=System.currentTimeMillis()

vardeferredZxing=GlobalScope.async{

getQrcodeInfoFromZxing(bitmap)

vardeferredZbar=GlobalScope.async{

getQrcodeInfoFromZbar(bitmap)

runBlocking{

//通過select監(jiān)聽zxing、zbar結(jié)果返回

varresult=selectString{

//監(jiān)聽zxing

deferredZxing.onAwait{value-

//value為deferredZxing識別的結(jié)果

"zxingresult$value"

//監(jiān)聽zbar

deferredZbar.onAwait{value-

"zbarresult$value"

//運行到此，說明已經(jīng)有結(jié)果返回

println("resultfrom$resultuseTime:${System.currentTimeMillis()-starTime}")

}

結(jié)果如下：

resultfromzbarresultImfishuseTime:1079

符合預(yù)期，同時可以看出：相比上個Demo，這樣寫簡潔了許多。

2.Select的使用

除了可以監(jiān)聽async的結(jié)果，Select還可以監(jiān)聽Channel的發(fā)送方/接收方數(shù)據(jù)，我們以監(jiān)聽接收方數(shù)據(jù)為例：

funtestSelect4(){

runBlocking{

varbitmap=null;

varstarTime=System.currentTimeMillis()

varreceiveChannelZxing=produce{

//生產(chǎn)數(shù)據(jù)

varresult=getQrcodeInfoFromZxing(bitmap)

//發(fā)送數(shù)據(jù)

send(result)

varreceiveChannelZbar=produce{

varresult=getQrcodeInfoFromZbar(bitmap)

send(result)

varresult=selectString{

//監(jiān)聽是否有數(shù)據(jù)發(fā)送過來

receiveChannelZxing.onReceive{

value-"zxingresult$value"

receiveChannelZbar.onReceive{

value-"zbarresult$value"

println("resultfrom$resultuseTime:${System.currentTimeMillis()-starTime}")

}

結(jié)果如下：

resultfromzbarresultImfishuseTime:1028

不論是async還是Channel，Select都可以監(jiān)聽它們的數(shù)據(jù)，從而形成多路復(fù)用的效果。

在監(jiān)聽協(xié)程里調(diào)用select表達式，表達式{}內(nèi)聲明需要監(jiān)聽的協(xié)程的數(shù)據(jù)，對于select來說有兩種場景：

沒有數(shù)據(jù)，則select掛起協(xié)程并等待直到其它協(xié)程數(shù)據(jù)準(zhǔn)備完成后再次恢復(fù)select所在的協(xié)程。有數(shù)據(jù)，則select正常執(zhí)行并返回獲取的數(shù)據(jù)。

3.Invoke函數(shù)的妙用

在分析Select原理之前，需要弄明白invoke函數(shù)的原理。

對于Kotlin類來說，都可以重寫其invoke函數(shù)。

operatorfuninvoke():String{

return"I'mfish"

如上，重寫了SelectDemo里的invoke函數(shù)，和普通成員函數(shù)一樣，我們可以通過對象調(diào)用它。

funmain(args:ArrayString){

varselectDemo=SelectDemo()

varresult=selectDemo.invoke()

println("result：$result")

當(dāng)然，可以進一步簡化：

funmain(args:ArrayString){

varselectDemo=SelectDemo()

varresult=selectDemo()

println("result：$result")

這里涉及到了kotlin的語法糖：對象居然可以像函數(shù)一樣調(diào)用。

作為函數(shù)，invoke當(dāng)然也可以接收高階函數(shù)作為參數(shù)：

operatorfuninvoke(block:(Int)-String):String{

returnblock(3)

funmain(args:ArrayString){

varselectDemo=SelectDemo()

varresult=selectDemo{age-

when(age){

3-"I'mfish3"

4-"I'mfish4"

else-"error"

println("result：$result")

因此，當(dāng)看到對象作為函數(shù)調(diào)用時，實際上調(diào)用的是invoke函數(shù)，具體的邏輯需要查看其invoke函數(shù)的實現(xiàn)。

4.Select的原理

上篇分析過Channel，因此本篇趁熱打鐵，通過Select監(jiān)聽Channel數(shù)據(jù)的變化來分析其原理，為方便講解，我們先以監(jiān)聽一個Channel的為例。

先從select表達式本身入手。

funtestSelect5(){

runBlocking{

varstarTime=System.currentTimeMillis()

varreceiveChannelZxing=produce{

//發(fā)送數(shù)據(jù)

send("I'mfish")

//確保channel數(shù)據(jù)已經(jīng)send

delay(1000)

varresult=selectString{

//監(jiān)聽是否有數(shù)據(jù)發(fā)送過來

receiveChannelZxing.onReceive{value-

"zxingresult$value"

println("resultfrom$resultuseTime:${System.currentTimeMillis()-starTime}")

}

select是掛起函數(shù)，因此協(xié)程運行到此有可能被掛起。

#Select.kt

publicsuspendinlinefunRselect(crossinlinebuilder:SelectBuilderR.()-Unit):R{

//...

returnsuspendCoroutineUninterceptedOrReturn{uCont-

//傳入父協(xié)程體

valscope=SelectBuilderImpl(uCont)

try{

//執(zhí)行builder

builder(scope)

}catch(e:Throwable){

scope.handleBuilderException(e)

//通過返回值判斷是否需要掛起協(xié)程

scope.getResult()

}

重點看builder(scope)，builder是高階函數(shù)，實際上就是執(zhí)行了select花括號里的內(nèi)容，而它里面就是監(jiān)聽數(shù)據(jù)是否返回。

receiveChannelZxing.onReceive

剛開始看的時候勢必以為onReceive是個函數(shù)，然而它是ReceiveChannel里的成員變量：

#Channel.kt

publicvalonReceive:SelectClause1E

通過上一節(jié)的分析可知，關(guān)鍵是要找到SelectClause1的invoke的實現(xiàn)。

#Select.kt

publicinterfaceSelectBuilderinR{

//block有個入?yún)?/p>

//聲明了SelectClause1的擴展函數(shù)invoke

publicoperatorfunQSelectClause1Q.invoke(block:suspend(Q)-R)

overridefunQSelectClause1Q.invoke(block:suspend(Q)-R){

//SelectBuilderImpl實現(xiàn)了SelectClause1的invoke函數(shù)

registerSelectClause1(this@SelectBuilderImpl,block)

}

再看onReceive的賦值：

#AbstractChannel.kt

finaloverridevalonReceive:SelectClause1E

get()=object:SelectClause1E{

@Suppress("UNCHECKED_CAST")

overridefunRregisterSelectClause1(select:SelectInstanceR,block:suspend(E)-R){

registerSelectReceiveMode(select,RECEIVE_THROWS_ON_CLOSE,blockassuspend(Any)-R)

因此，簡單總結(jié)調(diào)用棧如下：

當(dāng)調(diào)用receiveChannelZxing.onReceive{}，實際上調(diào)用了SelectClause1.invoke()，而它里面又調(diào)用了SelectClause1.registerSelectClause1()，最終調(diào)用了AbstractChannel.registerSelectReceiveMode。

AbstractChannel.registerSelectReceiveMode

#AbstractChannel.kt

privatefunRregisterSelectReceiveMode(select:SelectInstanceR,receiveMode:Int,block:suspend(Any)-R){

while(true){

//如果已經(jīng)有結(jié)果了，則直接返回-------①

if(select.isSelected)return

if(isEmptyImpl){

//沒有發(fā)送者在等待，則入隊等待，并返回-------②

if(enqueueReceiveSelect(select,block,receiveMode))return

}else{

//直接取出值-------③

valpollResult=pollSelectInternal(select)

when{

pollResult===ALREADY_SELECTED-return

pollResult===POLL_FAILED-{}//retry

pollResult===RETRY_ATOMIC-{}//retry

//調(diào)用block-------④

else-block.tryStartBlockUnintercepted(select,receiveMode,pollResult)

}

分為4個點，接著來一一分析。

①select同時監(jiān)聽多個值，若是有1個符合要求的數(shù)據(jù)返回了，那么該isSelected標(biāo)記為true，當(dāng)檢測到該標(biāo)記為true時直接退出。

結(jié)合之前的Demo，zbar已經(jīng)識別出結(jié)果了，當(dāng)select檢測zxing的結(jié)果時直接返回。

②:

#AbstractChannel.kt

privatefunRenqueueReceiveSelect(

select:SelectInstanceR,

block:suspend(Any)-R,

receiveMode:Int

):Boolean{

//構(gòu)造為Node元素

valnode=AbstractChannel.ReceiveSelect(this,select,block,receiveMode)

//添加到Channel隊列里

valresult=enqueueReceive(node)

if(result)select.disposeOnSelect(node)

returnresult

}

當(dāng)select時，發(fā)現(xiàn)Channel里沒有數(shù)據(jù)，說明Channel還沒有開始send，因此構(gòu)造了Node(ReceiveSele