【譯】kotlin 協程官方文檔（9）-選擇表達式(實驗階段)（Select Expression (experimental)）
- 2020 年 4 月 1 日
- 筆記
最近一直在了解關於kotlin協程的知識，那最好的學習資料自然是官方提供的學習文檔了，看了看後我就萌生了翻譯官方文檔的想法。前後花了要接近一個月時間，一共九篇文章，在這裡也分享出來，希望對讀者有所幫助。個人知識所限，有些翻譯得不是太順暢，也希望讀者能提出意見 協程官方文檔：coroutines-guide 協程官方文檔中文翻譯：coroutines-cn-guide 協程官方文檔中文譯者：leavesC
[TOC]
select 表達式可以同時等待多個掛起函數，並選擇第一個可用的函數來執行
選擇表達式是
kotlinx.coroutines的一個實驗性的特性，這些 API 預計將在kotlinx.coroutines庫的即將到來的更新中衍化，並可能會有突破性的變化
一、Selecting from channels
我們現在有兩個字元串生產者：fizz 和 buzz 。其中 fizz 每 300 毫秒生成一個字元串「Fizz」：
fun CoroutineScope.fizz() = produce<String> { while (true) { // sends "Fizz" every 300 ms delay(300) send("Fizz") } }
接著 buzz 每 500 毫秒生成一個字元串「Buzz!」：
fun CoroutineScope.buzz() = produce<String> { while (true) { // sends "Buzz!" every 500 ms delay(500) send("Buzz!") } }
使用掛起函數 receive，我們可以從兩個通道接收其中一個的數據。但是 select 表達式允許我們使用其 onReceive 子句同時從兩者接收：
suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) { select<Unit> { // <Unit> means that this select expression does not produce any result fizz.onReceive { value -> // this is the first select clause println("fizz -> '$value'") } buzz.onReceive { value -> // this is the second select clause println("buzz -> '$value'") } } }
讓我們運行程式碼 7 次：
import kotlinx.coroutines.* import kotlinx.coroutines.channels.* import kotlinx.coroutines.selects.* fun CoroutineScope.fizz() = produce<String> { while (true) { // sends "Fizz" every 300 ms delay(300) send("Fizz") } } fun CoroutineScope.buzz() = produce<String> { while (true) { // sends "Buzz!" every 500 ms delay(500) send("Buzz!") } } suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) { select<Unit> { // <Unit> means that this select expression does not produce any result fizz.onReceive { value -> // this is the first select clause println("fizz -> '$value'") } buzz.onReceive { value -> // this is the second select clause println("buzz -> '$value'") } } } fun main() = runBlocking<Unit> { //sampleStart val fizz = fizz() val buzz = buzz() repeat(7) { selectFizzBuzz(fizz, buzz) } coroutineContext.cancelChildren() // cancel fizz & buzz coroutines //sampleEnd }
運行結果：
fizz -> 'Fizz' buzz -> 'Buzz!' fizz -> 'Fizz' fizz -> 'Fizz' buzz -> 'Buzz!' fizz -> 'Fizz' buzz -> 'Buzz!'
二、Selecting on close
當通道關閉時，select 中的 onReceive 子句會失敗並導致相應的 select 引發異常。我們可以使用 onReceiveOrNull 子句在通道關閉時執行特定操作。下面的示例還顯示了 select 是一個返回其查詢方法結果的表達式：
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'" } }
注意，onReceiveOrNull 是一個擴展函數，僅可用於具有不可為空元素的通道，這樣就不會意外混淆通道是已關閉還是返回了空值這兩種情況
讓我們將其與生成四次「Hello」字元串的通道 a 和生成四次「World」字元串的通道 b 一起使用：
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) { // print first eight results println(selectAorB(a, b)) } coroutineContext.cancelChildren() //sampleEnd }
這段程式碼的結果非常有趣，所以我們將在細節中分析它：
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 偏向於第一個子句。當同時可以選擇多個子句時，將選擇其中的第一個子句。在這裡，兩個通道都在不斷地產生字元串，因此作為 select 中的第一個子句的通道獲勝。但是，因為我們使用的是無緩衝通道，所以 a 在其發送調用時會不時地被掛起，從而給了 b 發送的機會
第二個觀察結果是，當通道已經關閉時，onReceiveOrNull 將立即被選中
三、Selecting to send
select 表達式有 onSend 子句，可以與 selection 的偏向性質結合使用。 讓我們寫一個整數生產者的例子，當主通道上的消費者跟不上時，它會將其值發送到 side 通道：
fun CoroutineScope.produceNumbers(side: SendChannel<Int>) = produce<Int> { for (num in 1..10) { // produce 10 numbers from 1 to 10 delay(100) // every 100 ms select<Unit> { onSend(num) {} // Send to the primary channel side.onSend(num) {} // or to the side channel } } }
消費者將會非常緩慢，每個數值處理需要 250 毫秒：
import kotlinx.coroutines.* import kotlinx.coroutines.channels.* import kotlinx.coroutines.selects.* fun CoroutineScope.produceNumbers(side: SendChannel<Int>) = produce<Int> { for (num in 1..10) { // produce 10 numbers from 1 to 10 delay(100) // every 100 ms select<Unit> { onSend(num) {} // Send to the primary channel side.onSend(num) {} // or to the side channel } } } fun main() = runBlocking<Unit> { //sampleStart val side = Channel<Int>() // allocate side channel launch { // this is a very fast consumer for the side channel side.consumeEach { println("Side channel has $it") } } produceNumbers(side).consumeEach { println("Consuming $it") delay(250) // let us digest the consumed number properly, do not hurry } println("Done consuming") coroutineContext.cancelChildren() //sampleEnd }
讓我們看看會發生什麼：
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
四、Selecting deferred values
延遲值可以使用 onAwait 子句來查詢。讓我們啟動一個非同步函數，它在隨機的延遲後會延遲返回字元串：
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)) } }
現在，main 函數等待它們中的第一個完成，並統計仍處於活動狀態的延遲值的數量。注意，我們在這裡使用 select 表達式事實上是一種 Kotlin DSL，因此我們可以使用任意程式碼為它提供子句。在本例中，我們遍歷一個延遲值列表，為每個延遲值提供 onAwait 子句。
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 }
輸出結果：
Deferred 4 produced answer 'Waited for 128 ms' 11 coroutines are still active
五、Switch over a channel of deferred values
現在我們來編寫一個通道生產者函數，它消費一個產生延遲字元串的通道，並等待每個接收的延遲值，但它只在下一個延遲值到達或者通道關閉之前處於運行狀態。此示例將 onReceiveOrNull 和 onAwait 子句放在同一個 select 中：
fun CoroutineScope.switchMapDeferreds(input: ReceiveChannel<Deferred<String>>) = produce<String> { var current = input.receive() // start with first received deferred value while (isActive) { // loop while not cancelled/closed val next = select<Deferred<String>?> { // return next deferred value from this select or null input.onReceiveOrNull { update -> update // replaces next value to wait } current.onAwait { value -> send(value) // send value that current deferred has produced input.receiveOrNull() // and use the next deferred from the input channel } } if (next == null) { println("Channel was closed") break // out of loop } else { current = next } } }
為了測試它，我們將用一個簡單的非同步函數，它在特定的延遲後返回特定的字元串：
fun CoroutineScope.asyncString(str: String, time: Long) = async { delay(time) str }
main 函數只是啟動一個協程來列印 switchMapDeferreds 的結果並向它發送一些測試數據：
import kotlinx.coroutines.* import kotlinx.coroutines.channels.* import kotlinx.coroutines.selects.* fun CoroutineScope.switchMapDeferreds(input: ReceiveChannel<Deferred<String>>) = produce<String> { var current = input.receive() // start with first received deferred value while (isActive) { // loop while not cancelled/closed val next = select<Deferred<String>?> { // return next deferred value from this select or null input.onReceiveOrNull { update -> update // replaces next value to wait } current.onAwait { value -> send(value) // send value that current deferred has produced input.receiveOrNull() // and use the next deferred from the input channel } } if (next == null) { println("Channel was closed") break // out of loop } else { current = next } } } fun CoroutineScope.asyncString(str: String, time: Long) = async { delay(time) str } fun main() = runBlocking<Unit> { //sampleStart val chan = Channel<Deferred<String>>() // the channel for test launch { // launch printing coroutine for (s in switchMapDeferreds(chan)) println(s) // print each received string } chan.send(asyncString("BEGIN", 100)) delay(200) // enough time for "BEGIN" to be produced chan.send(asyncString("Slow", 500)) delay(100) // not enough time to produce slow chan.send(asyncString("Replace", 100)) delay(500) // give it time before the last one chan.send(asyncString("END", 500)) delay(1000) // give it time to process chan.close() // close the channel ... delay(500) // and wait some time to let it finish //sampleEnd }
程式碼的執行結果：
BEGIN Replace END Channel was closed
