Select表達式（實驗性） · Kotin知識點難點歸納總結

[TOC] ## select 表達式（實驗性的） select 表達式可以同時等待多個掛起函數，并 _選擇_第一個可用的。 > Select 表達式在 `kotlinx.coroutines` 中是一個實驗性的特性。這些 API 在`kotlinx.coroutines` 庫即將到來的更新中可能會發生改變。 ### 在通道中 select 我們現在有兩個字符串生產者：`fizz` 和 `buzz` 。其中 `fizz` 每 300 毫秒生成一個“Fizz”字符串： ```kotlin fun CoroutineScope.fizz() = produce<String> { while (true) { // 每 300 毫秒發送一個 "Fizz" delay(300) send("Fizz") } } ``` 接著 `buzz` 每 500 毫秒生成一個 “Buzz!” 字符串： ```kotlin fun CoroutineScope.buzz() = produce<String> { while (true) { // 每 500 毫秒發送一個"Buzz!" delay(500) send("Buzz!") } } ``` 使用 [receive](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-receive-channel/receive.html)掛起函數，我們可以從兩個通道接收 _其中一個_ 的數據。但是 [select](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.selects/select.html) 表達式允許我們使用其[onReceive](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-receive-channel/on-receive.html)子句 _同時_ 從兩者接收： ```kotlin suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) { select<Unit> { // <Unit> 意味著該 select 表達式不返回任何結果 fizz.onReceive { value -> // 這是第一個 select 子句 println("fizz -> '$value'") } buzz.onReceive { value -> // 這是第二個 select 子句 println("buzz -> '$value'") } } } ``` 讓我們運行代碼 7 次： ```kotlin import kotlinx.coroutines.* import kotlinx.coroutines.channels.* import kotlinx.coroutines.selects.* fun CoroutineScope.fizz() = produce<String> { while (true) { // 每 300 毫秒發送一個 "Fizz" delay(300) send("Fizz") } } fun CoroutineScope.buzz() = produce<String> { while (true) { // 每 500 毫秒發送一個 "Buzz!" delay(500) send("Buzz!") } } suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) { select<Unit> { // <Unit> 意味著該 select 表達式不返回任何結果 fizz.onReceive { value -> // 這是第一個 select 子句 println("fizz -> '$value'") } buzz.onReceive { value -> // 這是第二個 select 子句 println("buzz -> '$value'") } } } fun main() = runBlocking<Unit> { //sampleStart val fizz = fizz() val buzz = buzz() repeat(7) { selectFizzBuzz(fizz, buzz) } coroutineContext.cancelChildren() // 取消 fizz 和 buzz 協程 //sampleEnd } ``` > 可以在[這里](https://github.com/hltj/kotlinx.coroutines-cn/blob/master/kotlinx-coroutines-core/jvm/test/guide/example-select-01.kt)獲取完整代碼。這段代碼的執行結果如下： ```text fizz -> 'Fizz' buzz -> 'Buzz!' fizz -> 'Fizz' fizz -> 'Fizz' buzz -> 'Buzz!' fizz -> 'Fizz' buzz -> 'Buzz!' ``` ### 通道關閉時 select select 中的 [onReceive](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-receive-channel/on-receive.html)子句在已經關閉的通道執行會發生失敗，并導致相應的`select` 拋出異常。我們可以使用 [onReceiveOrNull](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/on-receive-or-null.html)子句在關閉通道時執行特定操作。以下示例還顯示了 `select` 是一個返回其查詢方法結果的表達式： ```kotlin suspend fun selectAorB(a: ReceiveChannel<String>, b: ReceiveChannel<String>): String = select<String> { a.onReceiveOrNull { value -> if (value == null) "Channel 'a' is closed" else "a -> '$value'" } b.onReceiveOrNull { value -> if (value == null) "Channel 'b' is closed" else "b -> '$value'" } } ``` Note that [onReceiveOrNull][onReceiveOrNull] is an extension function defined only for channels with non-nullable elements so that there is no accidental confusion between a closed channel and a null value. 現在有一個生成四次“Hello”字符串的 `a` 通道，和一個生成四次“World”字符串的 `b` 通道，我們在這兩個通道上使用它： ```kotlin import kotlinx.coroutines.* import kotlinx.coroutines.channels.* import kotlinx.coroutines.selects.* suspend fun selectAorB(a: ReceiveChannel<String>, b: ReceiveChannel<String>): String = select<String> { a.onReceiveOrNull { value -> if (value == null) "Channel 'a' is closed" else "a -> '$value'" } b.onReceiveOrNull { value -> if (value == null) "Channel 'b' is closed" else "b -> '$value'" } } fun main() = runBlocking<Unit> { //sampleStart val a = produce<String> { repeat(4) { send("Hello $it") } } val b = produce<String> { repeat(4) { send("World $it") } } repeat(8) { // 打印最早的八個結果 println(selectAorB(a, b)) } coroutineContext.cancelChildren() //sampleEnd } ``` > 可以在[這里](https://github.com/hltj/kotlinx.coroutines-cn/blob/master/kotlinx-coroutines-core/jvm/test/guide/example-select-02.kt)獲取完整代碼。這段代碼的結果非常有趣，所以我們將在細節中分析它： ```text a -> 'Hello 0' a -> 'Hello 1' b -> 'World 0' a -> 'Hello 2' a -> 'Hello 3' b -> 'World 1' Channel 'a' is closed Channel 'a' is closed ``` 有幾個結果可以通過觀察得出。首先，`select` _偏向于_ 第一個子句，當可以同時選到多個子句時，第一個子句將被選中。在這里，兩個通道都在不斷地生成字符串，因此 `a` 通道作為 select 中的第一個子句獲勝。然而因為我們使用的是無緩沖通道，所以 `a` 在其調用[send](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-send-channel/send.html)時會不時地被掛起，進而 `b` 也有機會發送。第二個觀察結果是，當通道已經關閉時，會立即選擇 [onReceiveOrNull](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/on-receive-or-null.html)。 ### Select 以發送 Select 表達式具有 [onSend](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-send-channel/on-send.html)子句，可以很好的與選擇的偏向特性結合使用。我們來編寫一個整數生成器的示例，當主通道上的消費者無法跟上它時，它會將值發送到 `side` 通道上： ```kotlin fun CoroutineScope.produceNumbers(side: SendChannel<Int>) = produce<Int> { for (num in 1..10) { // 生產從 1 到 10 的 10 個數值 delay(100) // 延遲 100 毫秒 select<Unit> { onSend(num) {} // 發送到主通道 side.onSend(num) {} // 或者發送到 side 通道 } } } ``` 消費者將會非常緩慢，每個數值處理需要 250 毫秒： ```kotlin import kotlinx.coroutines.* import kotlinx.coroutines.channels.* import kotlinx.coroutines.selects.* fun CoroutineScope.produceNumbers(side: SendChannel<Int>) = produce<Int> { for (num in 1..10) { // 生產從 1 到 10 的 10 個數值 delay(100) // 延遲 100 毫秒 select<Unit> { onSend(num) {} // 發送到主通道 side.onSend(num) {} // 或者發送到 side 通道 } } } fun main() = runBlocking<Unit> { //sampleStart val side = Channel<Int>() // 分配 side 通道 launch { // 對于 side 通道來說，這是一個很快的消費者 side.consumeEach { println("Side channel has $it") } } produceNumbers(side).consumeEach { println("Consuming $it") delay(250) // 不要著急，讓我們正確消化消耗被發送來的數字 } println("Done consuming") coroutineContext.cancelChildren() //sampleEnd } ``` > 可以在[這里](https://github.com/hltj/kotlinx.coroutines-cn/blob/master/kotlinx-coroutines-core/jvm/test/guide/example-select-03.kt)獲取完整代碼。讓我們看看會發生什么： ```text Consuming 1 Side channel has 2 Side channel has 3 Consuming 4 Side channel has 5 Side channel has 6 Consuming 7 Side channel has 8 Side channel has 9 Consuming 10 Done consuming ``` ### Select 延遲值延遲值可以使用 [onAwait](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-deferred/on-await.html)子句查詢。讓我們啟動一個異步函數，它在隨機的延遲后會延遲返回字符串： ```kotlin fun CoroutineScope.asyncString(time: Int) = async { delay(time.toLong()) "Waited for $time ms" } ``` 讓我們隨機啟動十余個異步函數，每個都延遲隨機的時間。 ```kotlin fun CoroutineScope.asyncStringsList(): List<Deferred<String>> { val random = Random(3) return List(12) { asyncString(random.nextInt(1000)) } } ``` 現在 main 函數在等待第一個函數完成，并統計仍處于激活狀態的延遲值的數量。注意，我們在這里使用 `select` 表達式事實上是作為一種 Kotlin DSL，所以我們可以用任意代碼為它提供子句。在這種情況下，我們遍歷一個延遲值的隊列，并為每個延遲值提供 `onAwait` 子句的調用。 ```kotlin import kotlinx.coroutines.* import kotlinx.coroutines.selects.* import java.util.* fun CoroutineScope.asyncString(time: Int) = async { delay(time.toLong()) "Waited for $time ms" } fun CoroutineScope.asyncStringsList(): List<Deferred<String>> { val random = Random(3) return List(12) { asyncString(random.nextInt(1000)) } } fun main() = runBlocking<Unit> { //sampleStart val list = asyncStringsList() val result = select<String> { list.withIndex().forEach { (index, deferred) -> deferred.onAwait { answer -> "Deferred $index produced answer '$answer'" } } } println(result) val countActive = list.count { it.isActive } println("$countActive coroutines are still active") //sampleEnd } ``` > 可以在[這里](https://github.com/hltj/kotlinx.coroutines-cn/blob/master/kotlinx-coroutines-core/jvm/test/guide/example-select-04.kt)獲取完整代碼。該輸出如下： ```text Deferred 4 produced answer 'Waited for 128 ms' 11 coroutines are still active ``` ### 在延遲值通道上切換我們現在來編寫一個通道生產者函數，它消費一個產生延遲字符串的通道，并等待每個接收的延遲值，但它只在下一個延遲值到達或者通道關閉之前處于運行狀態。此示例將[onReceiveOrNull](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/on-receive-or-null.html)和 [onAwait](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-deferred/on-await.html)子句放在同一個 `select` 中： ```kotlin fun CoroutineScope.switchMapDeferreds(input: ReceiveChannel<Deferred<String>>) = produce<String> { var current = input.receive() // 從第一個接收到的延遲值開始 while (isActive) { // 循環直到被取消或關閉 val next = select<Deferred<String>?> { // 從這個 select 中返回下一個延遲值或 null input.onReceiveOrNull { update -> update // 替換下一個要等待的值 } current.onAwait { value -> send(value) // 發送當前延遲生成的值 input.receiveOrNull() // 然后使用從輸入通道得到的下一個延遲值 } } if (next == null) { println("Channel was closed") break // 跳出循環 } else { current = next } } } ``` 為了測試它，我們將用一個簡單的異步函數，它在特定的延遲后返回特定的字符串： ```kotlin fun CoroutineScope.asyncString(str: String, time: Long) = async { delay(time) str } ``` main 函數只是啟動一個協程來打印 `switchMapDeferreds` 的結果并向它發送一些測試數據： ```kotlin import kotlinx.coroutines.* import kotlinx.coroutines.channels.* import kotlinx.coroutines.selects.* fun CoroutineScope.switchMapDeferreds(input: ReceiveChannel<Deferred<String>>) = produce<String> { var current = input.receive() // 從第一個接收到的延遲值開始 while (isActive) { // 循環直到被取消或關閉 val next = select<Deferred<String>?> { // 從這個 select 中返回下一個延遲值或 null input.onReceiveOrNull { update -> update // 替換下一個要等待的值 } current.onAwait { value -> send(value) // 發送當前延遲生成的值 input.receiveOrNull() // 然后使用從輸入通道得到的下一個延遲值 } } if (next == null) { println("Channel was closed") break // 跳出循環 } else { current = next } } } fun CoroutineScope.asyncString(str: String, time: Long) = async { delay(time) str } fun main() = runBlocking<Unit> { //sampleStart val chan = Channel<Deferred<String>>() // 測試使用的通道 launch { // 啟動打印協程 for (s in switchMapDeferreds(chan)) println(s) // 打印每個獲得的字符串 } chan.send(asyncString("BEGIN", 100)) delay(200) // 充足的時間來生產 "BEGIN" chan.send(asyncString("Slow", 500)) delay(100) // 不充足的時間來生產 "Slow" chan.send(asyncString("Replace", 100)) delay(500) // 在最后一個前給它一點時間 chan.send(asyncString("END", 500)) delay(1000) // 給執行一段時間 chan.close() // 關閉通道…… delay(500) // 然后等待一段時間來讓它結束 //sampleEnd } ``` > 可以在[這里](https://github.com/hltj/kotlinx.coroutines-cn/blob/master/kotlinx-coroutines-core/jvm/test/guide/example-select-05.kt)獲取完整代碼。這段代碼的執行結果： ```text BEGIN Replace END Channel was closed ``` [Deferred.onAwait]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-deferred/on-await.html [ReceiveChannel.receive]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-receive-channel/receive.html [ReceiveChannel.onReceive]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-receive-channel/on-receive.html [onReceiveOrNull]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/on-receive-or-null.html [SendChannel.send]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-send-channel/send.html [SendChannel.onSend]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-send-channel/on-send.html [select]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.selects/select.html