Kotlin?select使用方法介紹

更新時間：2022年11月24日 10:14:35 作者：且聽真言

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

一、select是什么

select——>用于選擇更快的結(jié)果。

基于場景理解

比如客戶端要查詢一個商品的詳情。兩個服務：緩存服務，速度快但信息可能是舊的；網(wǎng)絡服務，速度慢但信息一定是最新的。

如何實現(xiàn)上述邏輯：

   runBlocking {
        suspend fun getCacheInfo(productId: String): Product {
            delay(100L)
            return Product(productId, 8.9)
        }
        suspend fun getNetworkInfo(productId: String): Product? {
            delay(200L)
            return Product(productId, 8.8)
        }
        fun updateUI(product: Product) {
            println("${product.productId} : ${product.price}")
        }
        val startTime = System.currentTimeMillis()
        val productId = "001"
        val cacheInfo = getCacheInfo(productId)
        if (cacheInfo != null) {
            updateUI(cacheInfo)
            println("Time cost: ${System.currentTimeMillis() - startTime}")
        }
        val latestInfo = getNetworkInfo(productId)
        if (latestInfo != null) {
            updateUI(latestInfo)
            println("Time cost: ${System.currentTimeMillis() - startTime}")
        }
    }

001 : 8.9
Time cost: 113
001 : 8.8
Time cost: 324

上述程序分為四步：第一步：查詢緩存信息；第二步：緩存服務返回信息，更新 UI；第三步：查詢網(wǎng)絡服務；第四步：網(wǎng)絡服務返回信息，更新 UI。

用戶可以第一時間看到商品的信息，雖然它暫時會展示舊的信息，但由于我們同時查詢了網(wǎng)絡服務，舊緩存信息也馬上會被替代成新的信息。但是可能存在一些問題：如果程序卡在了緩存服務，獲取網(wǎng)絡服務就會無法執(zhí)行。是因為 getCacheInfo() 它是一個掛起函數(shù)，只有這個程序執(zhí)行成功以后，才可以繼續(xù)執(zhí)行后面的任務。能否做到：兩個掛起函數(shù)同時執(zhí)行，誰返回的速度更快，就選擇哪個結(jié)果。答案是使用select。

runBlocking {
        suspend fun getCacheInfo(productId: String): Product {
            delay(100L)
            return Product(productId, 8.9)
        }
        suspend fun getNetworkInfo(productId: String): Product {
            delay(200L)
            return Product(productId, 8.8)
        }
        fun updateUI(product: Product) {
            println("${product.productId} : ${product.price}")
        }
        val startTime = System.currentTimeMillis()
        val productId = "001"
        val product = select<Product?> {
            async {
                getCacheInfo(productId)
            }.onAwait {
                it
            }
            async {
                getNetworkInfo(productId)
            }.onAwait {
                it
            }
        }
        if (product != null) {
            updateUI(product)
            println("Time cost: ${System.currentTimeMillis() - startTime}")
        }
    }

001 : 8.9
Time cost: 134

Process finished with exit code 0

由于緩存的服務更快，所以，select 確實幫我們選擇了更快的那個結(jié)果。我們的 select 可以在緩存服務出現(xiàn)問題的時候，靈活選擇網(wǎng)絡服務的結(jié)果。從而避免用戶等待太長的時間，得到糟糕的體驗。

在上述代碼中，用戶大概率是會展示舊的緩存信息。但實際場景下，我們是需要進一步更新最新信息的。

runBlocking {
        suspend fun getCacheInfo(productId: String): Product {
            delay(100L)
            return Product(productId, 8.9)
        }
        suspend fun getNetworkInfo(productId: String): Product {
            delay(200L)
            return Product(productId, 8.8)
        }
        fun updateUI(product: Product) {
            println("${product.productId} : ${product.price}")
        }
        val startTime = System.currentTimeMillis()
        val productId = "001"
        val cacheDeferred = async {
            getCacheInfo(productId)
        }
        val latestDeferred = async {
            getNetworkInfo(productId)
        }
        val product = select<Product?> {
 
            cacheDeferred.onAwait {
                it.copy(isCache = true)
            }
            latestDeferred.onAwait {
                it.copy(isCache = false)
            }
        }
        if (product != null) {
            updateUI(product)
            println("Time cost: ${System.currentTimeMillis() - startTime}")
        }
        if (product != null && product.isCache) {
            val latest = latestDeferred.await() ?: return@runBlocking
            updateUI(latest)
            println("Time cost: ${System.currentTimeMillis() - startTime}")
        }
    }

001 : 8.9
Time cost: 124
001 : 8.8
Time cost: 228

Process finished with exit code 0

如果是多個服務的緩存場景呢？

 runBlocking {
        val startTime = System.currentTimeMillis()
        val productId = "001"
        suspend fun getCacheInfo(productId: String): Product {
            delay(100L)
            return Product(productId, 8.9)
        }
        suspend fun getCacheInfo2(productId: String): Product {
            delay(50L)
            return Product(productId, 8.85)
        }
        suspend fun getNetworkInfo(productId: String): Product {
            delay(200L)
            return Product(productId, 8.8)
        }
        fun updateUI(product: Product) {
            println("${product.productId} : ${product.price}")
        }
        val cacheDeferred = async {
            getCacheInfo(productId)
        }
        val cacheDeferred2 = async {
            getCacheInfo2(productId)
        }
        val latestDeferred = async {
            getNetworkInfo(productId)
        }
        val product = select<Product?> {
            cacheDeferred.onAwait {
                it.copy(isCache = true)
            }
            cacheDeferred2.onAwait {
                it.copy(isCache = true)
            }
            latestDeferred.onAwait {
                it.copy(isCache = true)
            }
        }
        if (product != null) {
            updateUI(product)
            println("Time cost: ${System.currentTimeMillis() - startTime}")
        }
        if (product != null && product.isCache) {
            val latest = latestDeferred.await()
            updateUI(latest)
            println("Time cost: ${System.currentTimeMillis() - startTime}")
        }
    }

Log

001 : 8.85
Time cost: 79
001 : 8.8
Time cost: 229

Process finished with exit code 0

select 代碼模式，可以提升程序的整體響應速度。

二、select和Channel

runBlocking {
        val startTime = System.currentTimeMillis()
        val channel1 = produce {
            send(1)
            delay(200L)
            send(2)
            delay(200L)
            send(3)
        }
        val channel2 = produce {
            delay(100L)
            send("a")
            delay(200L)
            send("b")
            delay(200L)
            send("c")
        }
        channel1.consumeEach {
            println(it)
        }
        channel2.consumeEach {
            println(it)
        }
        println("Time cost: ${System.currentTimeMillis() - startTime}")
    }

Log

1
2
3
a
b
c
Time cost: 853

Process finished with exit code 0

上述代碼串行執(zhí)行，可以使用select進行優(yōu)化。

  runBlocking {
        val startTime = System.currentTimeMillis()
        val channel1 = produce {
            send(1)
            delay(200L)
            send(2)
            delay(200L)
            send(3)
        }
        val channel2 = produce {
            delay(100L)
            send("a")
            delay(200L)
            send("b")
            delay(200L)
            send("c")
        }
        suspend fun selectChannel(
            channel1: ReceiveChannel<Int>,
            channel2: ReceiveChannel<String>
        ): Any {
            return select<Any> {
                if (!channel1.isClosedForReceive) {
                    channel1.onReceive {
                        it.also {
                            println(it)
                        }
                    }
                }
                if (!channel2.isClosedForReceive) {
                    channel2.onReceive {
                        it.also {
                            println(it)
                        }
                    }
                }
            }
        }
        repeat(6) {
            selectChannel(channel1, channel2)
        }
        println("Time cost: ${System.currentTimeMillis() - startTime}")
    }

Log
1
a
2
b
3
c
Time cost: 574

Process finished with exit code 0

從代碼執(zhí)行結(jié)果可以發(fā)現(xiàn)程序的執(zhí)行耗時有效減少。onReceive{} 是 Channel 在 select 當中的語法，當 Channel 當中有數(shù)據(jù)以后，它就會被回調(diào)，通過這個 Lambda，將結(jié)果傳出去。執(zhí)行了 6 次 select，目的是要把兩個管道中的所有數(shù)據(jù)都消耗掉。

如果Channel1不生產(chǎn)數(shù)據(jù)了，程序會如何執(zhí)行？

runBlocking {
        val startTime = System.currentTimeMillis()
        val channel1 = produce<String> {
            delay(5000L)
        }
        val channel2 = produce<String> {
            delay(100L)
            send("a")
            delay(200L)
            send("b")
            delay(200L)
            send("c")
        }
        suspend fun selectChannel(
            channel1: ReceiveChannel<String>,
            channel2: ReceiveChannel<String>
        ): String = select<String> {
            channel1.onReceive {
                it.also {
                    println(it)
                }
            }
            channel2.onReceive {
                it.also {
                    println(it)
                }
            }
        }
        repeat(3) {
            selectChannel(channel1, channel2)
        }
        println("Time cost: ${System.currentTimeMillis() - startTime}")
    }

Log
a
b
c
Time cost: 570

Process finished with exit code 0

如果不知道Channel的個數(shù)，如何避免ClosedReceiveChannelException?

使用：onReceiveCatching{}

runBlocking {
        val startTime = System.currentTimeMillis()
        val channel1 = produce<String> {
            delay(5000L)
        }
        val channel2 = produce<String> {
            delay(100L)
            send("a")
            delay(200L)
            send("b")
            delay(200L)
            send("c")
        }
        suspend fun selectChannel(
            channel1: ReceiveChannel<String>,
            channel2: ReceiveChannel<String>
        ): String = select<String> {
            channel1.onReceiveCatching {
                it.getOrNull() ?: "channel1 is closed!"
            }
            channel2.onReceiveCatching {
                it.getOrNull() ?: "channel2 is closed!"
            }
        }
        repeat(6) {
            val result = selectChannel(channel1, channel2)
            println(result)
        }
        println("Time cost: ${System.currentTimeMillis() - startTime}")
    }

Log
a
b
c
channel2 is closed!
channel2 is closed!
channel2 is closed!
Time cost: 584

Process finished with exit code 0

得到所有結(jié)果以后，程序不會立即退出，因為 channel1 一直在 delay()。

所以我們需要在6次repeat之后將channel關(guān)閉。

runBlocking {
        val startTime = System.currentTimeMillis()
        val channel1 = produce<String> {
            delay(15000L)
        }
        val channel2 = produce<String> {
            delay(100L)
            send("a")
            delay(200L)
            send("b")
            delay(200L)
            send("c")
        }
        suspend fun selectChannel(
            channel1: ReceiveChannel<String>,
            channel2: ReceiveChannel<String>
        ): String = select<String> {
            channel1.onReceiveCatching {
                it.getOrNull() ?: "channel1 is closed!"
            }
            channel2.onReceiveCatching {
                it.getOrNull() ?: "channel2 is closed!"
            }
        }
        repeat(6) {
            val result = selectChannel(channel1, channel2)
            println(result)
        }
        channel1.cancel()
        channel2.cancel()
        println("Time cost: ${System.currentTimeMillis() - startTime}")
    }

Log
a
b
c
channel2 is closed!
channel2 is closed!
channel2 is closed!
Time cost: 612

Process finished with exit code 0

Deferred、Channel 的 API:

public interface Deferred : CoroutineContext.Element {
    public suspend fun join()
    public suspend fun await(): T
    public val onJoin: SelectClause0
    public val onAwait: SelectClause1<T>
}
public interface SendChannel<in E> 
    public suspend fun send(element: E)
    public val onSend: SelectClause2<E, SendChannel<E>>
}
public interface ReceiveChannel<out E> {
    public suspend fun receive(): E
    public suspend fun receiveCatching(): ChannelResult<E>
    public val onReceive: SelectClause1<E>
    public val onReceiveCatching: SelectClause1<ChannelResult<E>>
}

當 select 與 Deferred 結(jié)合使用的時候，當并行的 Deferred 比較多的時候，你往往需要在得到一個最快的結(jié)果以后，去取消其他的 Deferred。

通過 async 并發(fā)執(zhí)行協(xié)程，也可以借助 select 得到最快的結(jié)果。

 runBlocking {
        suspend fun <T> fastest(vararg deferreds: Deferred<T>): T = select {
            fun cancelAll() = deferreds.forEach {
                it.cancel()
            }
            for (deferred in deferreds) {
                deferred.onAwait {
                    cancelAll()
                    it
                }
            }
        }
        val deferred1 = async {
            delay(100L)
            println("done1")
            "result1"
        }
        val deferred2 = async {
            delay(200L)
            println("done2")
            "result2"
        }
        val deferred3 = async {
            delay(300L)
            println("done3")
            "result3"
        }
        val deferred4 = async {
            delay(400L)
            println("done4")
            "result4"
        }
        val deferred5 = async {
            delay(5000L)
            println("done5")
            "result5"
        }
        val fastest = fastest(deferred1, deferred2, deferred3, deferred4, deferred5)
        println(fastest)
    }