【譯】kotlin 協程官方文檔(9)-選擇表達式(實驗階段)(Select Expression (experimental))

最近一直在瞭解關於kotlin協程的知識,那最好的學習資料天然是官方提供的學習文檔了,看了看後我就萌生了翻譯官方文檔的想法。先後花了要接近一個月時間,一共九篇文章,在這裏也分享出來,但願對讀者有所幫助。我的知識所限,有些翻譯得不是太順暢,也但願讀者能提出意見java

協程官方文檔:coroutines-guidegit

協程官方文檔中文翻譯:coroutines-cn-guidegithub

協程官方文檔中文譯者:leavesCexpress

[TOC]dom

select 表達式能夠同時等待多個掛起函數,並選擇第一個可用的函數來執行異步

選擇表達式是 kotlinx.coroutines 的一個實驗性的特性,這些 API 預計將在 kotlinx.coroutines 庫的即將到來的更新中衍化,並可能會有突破性的變化async

1、Selecting from channels

咱們如今有兩個字符串生產者:fizzbuzz 。其中 fizz 每 300 毫秒生成一個字符串「Fizz」:ide

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 子句同時從二者接收:oop

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!'
複製代碼

2、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 將當即被選中

3、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
複製代碼

4、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
複製代碼

5、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
複製代碼
相關文章
相關標籤/搜索