Kotlin Select協(xié)程多路復(fù)用的實(shí)現(xiàn)詳解

更新時(shí)間：2022年09月13日 15:23:34 作者：小魚人愛編程

select是Kotlin 1.6中的特性，即選擇最快的結(jié)果。select與async、Channel結(jié)合使用，可以大大提高程序的響應(yīng)速度，還可以提高程序的靈活性、擴(kuò)展性

前言

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

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

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

1. Select 的引入

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

串行執(zhí)行

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

如下：

    //從zxing 獲取二維碼信息
    suspend fun getQrcodeInfoFromZxing(bitmap: Bitmap?): String {
        //模擬耗時(shí)
        delay(2000)
        return "I'm fish"
    }
    //從zbar 獲取二維碼信息
    suspend fun getQrcodeInfoFromZbar(bitmap: Bitmap?): String {
        delay(1000)
        return "I'm fish"
    }
    fun testSelect() {
        runBlocking {
            var bitmap = null
            var starTime = System.currentTimeMillis()
            var qrcoe1 = getQrcodeInfoFromZxing(bitmap)
            var qrcode2 = getQrcodeInfoFromZbar(bitmap)
            println("qrcode1=$qrcoe1 qrcode2=$qrcode2 useTime:${System.currentTimeMillis() - starTime} ms")
        }
    }

查看打印，最后花費(fèi)的時(shí)間：

qrcode1=I’m fish qrcode2=I’m fish useTime:3013 ms

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

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

如下：

    fun testSelect1() {
        var bitmap = null;
        var starTime = System.currentTimeMillis()
        var deferredZxing = GlobalScope.async {
            getQrcodeInfoFromZxing(bitmap)
        }
        var deferredZbar = GlobalScope.async {
            getQrcodeInfoFromZbar(bitmap)
        }
        runBlocking {
            //掛起等待識別結(jié)果
            var qrcoe1 = deferredZxing.await()
            //掛起等待識別結(jié)果
            var qrcode2 = deferredZbar.await()
            println("qrcode1=$qrcoe1 qrcode2=$qrcode2 useTime:${System.currentTimeMillis() - starTime} ms")
        }
    }

查看打印，最后花費(fèi)的時(shí)間：

qrcode1=I’m fish qrcode2=I’m fish useTime:2084 ms

可以看出，花費(fèi)時(shí)間明顯變少了。

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

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

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

如下：

    fun testSelect2() {
        var bitmap = null;
        var starTime = System.currentTimeMillis()
        var deferredZxing = GlobalScope.async {
            getQrcodeInfoFromZxing(bitmap)
        }
        var deferredZbar = GlobalScope.async {
            getQrcodeInfoFromZbar(bitmap)
        }
        var isEnd = false
        var result: String? = null
        GlobalScope.launch {
            if (!isEnd) {
                //沒有結(jié)束，則繼續(xù)識別
                var resultTmp = deferredZxing.await()
                if (!isEnd) {
                    //識別沒有結(jié)束，說明自己是第一個(gè)返回結(jié)果的
                    result = resultTmp
                    println("zxing recognize ok useTime:${System.currentTimeMillis() - starTime} ms")
                    //標(biāo)記識別結(jié)束
                    isEnd = true
                }
            }
        }
        GlobalScope.launch {
            if (!isEnd) {
                var resultTmp = deferredZbar.await()
                if (!isEnd) {
                    //識別沒有結(jié)束，說明自己是第一個(gè)返回結(jié)果的
                    result = resultTmp
                    println("zbar recognize ok useTime:${System.currentTimeMillis() - starTime} ms")
                    isEnd = true
                }
            }
        }
        //檢測是否有結(jié)果返回
        runBlocking {
            while (!isEnd) {
                delay(1)
            }
            println("recognize result:$result")
        }
    }

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

結(jié)果如下：

zbar recognize ok useTime:1070 ms

recognize result:I’m fish

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

Select 閃亮登場

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

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

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

    fun testSelect3() {
        var bitmap = null;
        var starTime = System.currentTimeMillis()
        var deferredZxing = GlobalScope.async {
            getQrcodeInfoFromZxing(bitmap)
        }
        var deferredZbar = GlobalScope.async {
            getQrcodeInfoFromZbar(bitmap)
        }
        runBlocking {
            //通過select 監(jiān)聽zxing、zbar 結(jié)果返回
            var result = select<String> {
                //監(jiān)聽zxing
                deferredZxing.onAwait {value->
                    //value 為deferredZxing 識別的結(jié)果
                    "zxing result $value"
                }
                //監(jiān)聽zbar
                deferredZbar.onAwait { value->
                    "zbar result $value"
                }
            }
            //運(yùn)行到此，說明已經(jīng)有結(jié)果返回
            println("result from $result useTime:${System.currentTimeMillis() - starTime}")
        }
    }

結(jié)果如下：

result from zbar result I’m fish useTime:1079

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

2. Select 的使用

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

    fun testSelect4() {
        runBlocking {
            var bitmap = null;
            var starTime = System.currentTimeMillis()
            var receiveChannelZxing = produce {
                //生產(chǎn)數(shù)據(jù)
                var result = getQrcodeInfoFromZxing(bitmap)
                //發(fā)送數(shù)據(jù)
                send(result)
            }
            var receiveChannelZbar = produce {
                var result = getQrcodeInfoFromZbar(bitmap)
                send(result)
            }
            var result = select<String> {
                //監(jiān)聽是否有數(shù)據(jù)發(fā)送過來
                receiveChannelZxing.onReceive {
                    value->"zxing result $value"
                }
                receiveChannelZbar.onReceive {
                        value->"zbar result $value"
                }
            }
            println("result from $result useTime:${System.currentTimeMillis() - starTime}")
        }
    }

結(jié)果如下：

result from zbar result I’m fish useTime:1028

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

在監(jiān)聽協(xié)程里調(diào)用select 表達(dá)式，表達(dá)式{}內(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ù)。

    operator fun invoke():String {
        return "I'm fish"
    }

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

fun main(args: Array<String>) {
    var selectDemo = SelectDemo()
    var result = selectDemo.invoke()
    println("result：$result")
}

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

fun main(args: Array<String>) {
    var selectDemo = SelectDemo()
    var result = selectDemo()
    println("result：$result")
}

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

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

    operator fun invoke(block: (Int) -> String): String {
        return block(3)
    }
fun main(args: Array<String>) {
    var selectDemo = SelectDemo()
    var result = selectDemo { age ->
        when (age) {
            3 -> "I'm fish3"
            4 -> "I'm fish4"
            else -> "error"
        }
    }
    println("result：$result")
}

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

4. Select 的原理

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

先從select 表達(dá)式本身入手。

    fun testSelect5() {
        runBlocking {
            var starTime = System.currentTimeMillis()
            var receiveChannelZxing = produce {
                //發(fā)送數(shù)據(jù)
                send("I'm fish")
            }
            //確保channel 數(shù)據(jù)已經(jīng)send
            delay(1000)
            var result = select<String> {
                //監(jiān)聽是否有數(shù)據(jù)發(fā)送過來
                receiveChannelZxing.onReceive { value ->
                    "zxing result $value"
                }
            }
            println("result from $result useTime:${System.currentTimeMillis() - starTime}")
        }
    }

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

#Select.kt
public suspend inline fun <R> select(crossinline builder: SelectBuilder<R>.() -> Unit): R {
    //...
    return suspendCoroutineUninterceptedOrReturn { uCont ->
        //傳入父協(xié)程體
        val scope = SelectBuilderImpl(uCont)
        try {
            //執(zhí)行builder
            builder(scope)
        } catch (e: Throwable) {
            scope.handleBuilderException(e)
        }
        //通過返回值判斷是否需要掛起協(xié)程
        scope.getResult()
    }
}

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

receiveChannelZxing.onReceive

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

#Channel.kt
    public val onReceive: SelectClause1<E>

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

#Select.kt
public interface SelectBuilder<in R> {
    //block 有個(gè)入?yún)?
    //聲明了SelectClause1的擴(kuò)展函數(shù)invoke
    public operator fun <Q> SelectClause1<Q>.invoke(block: suspend (Q) -> R)
}
override fun <Q> SelectClause1<Q>.invoke(block: suspend (Q) -> R) {
    //SelectBuilderImpl 實(shí)現(xiàn)了 SelectClause1 的invoke函數(shù)
    registerSelectClause1(this@SelectBuilderImpl, block)
}

再看onReceive 的賦值：

#AbstractChannel.kt
final override val onReceive: SelectClause1<E>
    get() = object : SelectClause1<E> {
        @Suppress("UNCHECKED_CAST")
        override fun <R> registerSelectClause1(select: SelectInstance<R>, block: suspend (E) -> R) {
            registerSelectReceiveMode(select, RECEIVE_THROWS_ON_CLOSE, block as suspend (Any?) -> R)
        }
    }

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

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

AbstractChannel. registerSelectReceiveMode

#AbstractChannel.kt
private fun <R> registerSelectReceiveMode(select: SelectInstance<R>, receiveMode: Int, block: suspend (Any?) -> R) {
    while (true) {
        //如果已經(jīng)有結(jié)果了，則直接返回------->①
        if (select.isSelected) return
        if (isEmptyImpl) {
            //沒有發(fā)送者在等待，則入隊(duì)等待，并返回 ------->②
            if (enqueueReceiveSelect(select, block, receiveMode)) return
        } else {
            //直接取出值------->③
            val pollResult = 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個(gè)點(diǎn)，接著來一一分析。

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

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

②:

#AbstractChannel.kt
private fun <R> enqueueReceiveSelect(
    select: SelectInstance<R>,
    block: suspend (Any?) -> R,
    receiveMode: Int
): Boolean {
    //構(gòu)造為Node元素
    val node = AbstractChannel.ReceiveSelect(this, select, block, receiveMode)
    //添加到Channel隊(duì)列里
    val result = enqueueReceive(node)
    if (result) select.disposeOnSelect(node)
    return result
}

當(dāng)select 時(shí)，發(fā)現(xiàn)Channel里沒有數(shù)據(jù)，說明Channel還沒有開始send，因此構(gòu)造了Node(ReceiveSelect)加入到Channel queue里。當(dāng)send數(shù)據(jù)時(shí)，會(huì)查找queue里是否有接收者等待，若有則調(diào)用Node(ReceiveSelect.completeResumeReceive)：

#AbstractChannel.kt
        override fun completeResumeReceive(value: E) {
            block.startCoroutineCancellable(
                if (receiveMode == RECEIVE_RESULT) ChannelResult.success(value) else value,
                select.completion,
                resumeOnCancellationFun(value)
            )
        }

block 被調(diào)度執(zhí)行，最后會(huì)恢復(fù)select 協(xié)程的執(zhí)行。

③取出數(shù)據(jù)，并嘗試恢復(fù)send協(xié)程。

④在③的基礎(chǔ)上，拿到數(shù)據(jù)后，直接執(zhí)行block（此時(shí)并沒有切換線程進(jìn)行調(diào)度）。

小結(jié)一下select 原理：

可以看出：

select 本身執(zhí)行并不耗時(shí)，若最終沒有數(shù)據(jù)返回則掛起等待，若是有數(shù)據(jù)返回則不會(huì)掛起協(xié)程。

我們從頭再捋一下select 配合Channel 的原理：

雖然以Channel為例講解了select 原理，實(shí)際上async等結(jié)合select 原理大致差不多，重點(diǎn)都是利用了協(xié)程的掛起/恢復(fù)做文章。

5. Select注意事項(xiàng)

如果select有多個(gè)數(shù)據(jù)同時(shí)到達(dá)，select 默認(rèn)會(huì)選擇第一個(gè)數(shù)據(jù)，若想要隨機(jī)選擇數(shù)據(jù)，可做如下處理：

            var result = selectUnbiased<String> {
                //監(jiān)聽是否有數(shù)據(jù)發(fā)送過來
                receiveChannelZxing.onReceive { value ->
                    "zxing result $value"
                }
            }

想要知道select 還可以監(jiān)聽哪些數(shù)據(jù)，可查看該數(shù)據(jù)是否實(shí)現(xiàn)了SelectClauseX(X 表示0、1、2）。

以上即為Select 的原理及其使用，下篇將會(huì)進(jìn)入?yún)f(xié)程的精華部分：Flow的運(yùn)用，該部分內(nèi)容較多，可能會(huì)分幾篇分析，敬請期待。

本文基于Kotlin 1.5.3，文中完整Demo傳送門

到此這篇關(guān)于Kotlin Select多路復(fù)用的實(shí)現(xiàn)詳解的文章就介紹到這了,更多相關(guān)Kotlin Select 內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: