第KotlinSelect協程多路復用的實現詳解目錄前言1.Select的引入多路數據的選擇串行執行協程并行執行同時監聽多路結果Select閃亮登場2.Select的使用3.Invoke函數的妙用4.Select的原理5.Select注意事項

前言

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

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

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

1.Select的引入

多路數據的選擇

串行執行

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

如下：

//從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

當然這是串行的方式效率比較低，我們想到了用協程來優化它。

協程并行執行

如下：

funtestSelect1(){

varbitmap=null;

varstarTime=System.currentTimeMillis()

vardeferredZxing=GlobalScope.async{

getQrcodeInfoFromZxing(bitmap)

vardeferredZbar=GlobalScope.async{

getQrcodeInfoFromZbar(bitmap)

runBlocking{

//掛起等待識別結果

varqrcoe1=deferredZxing.await()

//掛起等待識別結果

varqrcode2=deferredZbar.await()

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

}

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

qrcode1=Imfishqrcode2=ImfishuseTime:2084ms

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

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

同時監聽并記錄識別結果的返回。

同時監聽多路結果

如下：

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){

//沒有結束，則繼續識別

varresultTmp=deferredZxing.await()

if(!isEnd){

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

result=resultTmp

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

//標記識別結束

isEnd=true

GlobalScope.launch{

if(!isEnd){

varresultTmp=deferredZbar.await()

if(!isEnd){

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

result=resultTmp

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

isEnd=true

//檢測是否有結果返回

runBlocking{

while(!isEnd){

delay(1)

println("recognizeresult:$result")

}

通過檢測isEnd標記來判斷是否有某個模塊返回結果。

結果如下：

zbarrecognizeokuseTime:1070ms

recognizeresult:Imfish

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

Select閃亮登場

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

Select就是為了解決多路數據的選擇而生的。

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

funtestSelect3(){

varbitmap=null;

varstarTime=System.currentTimeMillis()

vardeferredZxing=GlobalScope.async{

getQrcodeInfoFromZxing(bitmap)

vardeferredZbar=GlobalScope.async{

getQrcodeInfoFromZbar(bitmap)

runBlocking{

//通過select監聽zxing、zbar結果返回

varresult=selectString{

//監聽zxing

deferredZxing.onAwait{value-

//value為deferredZxing識別的結果

"zxingresult$value"

//監聽zbar

deferredZbar.onAwait{value-

"zbarresult$value"

//運行到此，說明已經有結果返回

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

}

結果如下：

resultfromzbarresultImfishuseTime:1079

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

2.Select的使用

除了可以監聽async的結果，Select還可以監聽Channel的發送方/接收方數據，我們以監聽接收方數據為例：

funtestSelect4(){

runBlocking{

varbitmap=null;

varstarTime=System.currentTimeMillis()

varreceiveChannelZxing=produce{

//生產數據

varresult=getQrcodeInfoFromZxing(bitmap)

//發送數據

send(result)

varreceiveChannelZbar=produce{

varresult=getQrcodeInfoFromZbar(bitmap)

send(result)

varresult=selectString{

//監聽是否有數據發送過來

receiveChannelZxing.onReceive{

value-"zxingresult$value"

receiveChannelZbar.onReceive{

value-"zbarresult$value"

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

}

結果如下：

resultfromzbarresultImfishuseTime:1028

不論是async還是Channel，Select都可以監聽它們的數據，從而形成多路復用的效果。

在監聽協程里調用select表達式，表達式{}內聲明需要監聽的協程的數據，對于select來說有兩種場景：

沒有數據，則select掛起協程并等待直到其它協程數據準備完成后再次恢復select所在的協程。有數據，則select正常執行并返回獲取的數據。

3.Invoke函數的妙用

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

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

operatorfuninvoke():String{

return"I'mfish"

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

funmain(args:ArrayString){

varselectDemo=SelectDemo()

varresult=selectDemo.invoke()

println("result：$result")

當然，可以進一步簡化：

funmain(args:ArrayString){

varselectDemo=SelectDemo()

varresult=selectDemo()

println("result：$result")

這里涉及到了kotlin的語法糖：對象居然可以像函數一樣調用。

作為函數，invoke當然也可以接收高階函數作為參數：

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")

因此，當看到對象作為函數調用時，實際上調用的是invoke函數，具體的邏輯需要查看其invoke函數的實現。

4.Select的原理

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

先從select表達式本身入手。

funtestSelect5(){

runBlocking{

varstarTime=System.currentTimeMillis()

varreceiveChannelZxing=produce{

//發送數據

send("I'mfish")

//確保channel數據已經send

delay(1000)

varresult=selectString{

//監聽是否有數據發送過來

receiveChannelZxing.onReceive{value-

"zxingresult$value"

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

}

select是掛起函數，因此協程運行到此有可能被掛起。

#Select.kt

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

//...

returnsuspendCoroutineUninterceptedOrReturn{uCont-

//傳入父協程體

valscope=SelectBuilderImpl(uCont)

try{

//執行builder

builder(scope)

}catch(e:Throwable){

scope.handleBuilderException(e)

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

scope.getResult()

}

重點看builder(scope)，builder是高階函數，實際上就是執行了select花括號里的內容，而它里面就是監聽數據是否返回。

receiveChannelZxing.onReceive

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

#Channel.kt

publicvalonReceive:SelectClause1E

通過上一節的分析可知，關鍵是要找到SelectClause1的invoke的實現。

#Select.kt

publicinterfaceSelectBuilderinR{

//block有個入參

//聲明了SelectClause1的擴展函數invoke

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

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

//SelectBuilderImpl實現了SelectClause1的invoke函數

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)

因此，簡單總結調用棧如下：

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

AbstractChannel.registerSelectReceiveMode

#AbstractChannel.kt

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

while(true){

//如果已經有結果了，則直接返回-------①

if(select.isSelected)return

if(isEmptyImpl){

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

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

}else{

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

valpollResult=pollSelectInternal(select)

when{

pollResult===ALREADY_SELECTED-return

pollResult===POLL_FAILED-{}//retry

pollResult===RETRY_ATOMIC-{}//retry

//調用block-------④

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

}

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

①select同時監聽多個值，若是有1個符合要求的數據返回了，那么該isSelected標記為true，當檢測到該標記為true時直接退出。

結合之前的Demo，zbar已經識別出結果了，當select檢測zxing的結果時直接返回。

②:

#AbstractChannel.kt

privatefunRenqueueReceiveSelect(

select:SelectInstanceR,

block:suspend(Any)-R,

receiveMode:Int

):Boolean{

//構造為Node元素

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

//添加到Channel隊列里

valresult=enqueueReceive(node)

if(result)select.disposeOnSelect(node)

returnresult

}

當select時，發現Channel里沒有數據，說明Channel還沒有開始send，因此構造了Node(ReceiveSele