Kotlin 语言参考文档 中文版 Help

选择表达式(Select expression) (实验性功能)

使用选择表达式, 我们可以同时等待多个挂起函数, 并且 选择 其中第一个执行完毕的结果.

从通道中选择

假设我们有两个 string 值的生产者: fizzbuzz. 其中 fizz 每 500ms 产生一个 "Fizz" 字符串:

fun CoroutineScope.fizz() = produce<String> { while (true) { // 每 500ms 发送一个 "Fizz" delay(500) send("Fizz") } }

buzz 每 1000ms 产生一个 "Buzz!" 字符串:

fun CoroutineScope.buzz() = produce<String> { while (true) { // 每 1000ms 发生一个 "Buzz!" delay(1000) send("Buzz!") } }

使用 receive 挂起函数, 我们可以接收这两个通道中的 任何一个. 但使用 select 表达式的 onReceive 子句, 我们可以 同时接收两个通道的数据:

suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) { select<Unit> { // <Unit> 表示这个 select 表达式不产生任何结果值 fizz.onReceive { value -> // 这是第 1 个 select 子句 println("fizz -> '$value'") } buzz.onReceive { value -> // 这是第 2 个 select 子句 println("buzz -> '$value'") } } }

下面我们把这段代码运行 7 次:

import kotlinx.coroutines.* import kotlinx.coroutines.channels.* import kotlinx.coroutines.selects.* fun CoroutineScope.fizz() = produce<String> { while (true) { // 每 500ms 发送一个 "Fizz" delay(500) send("Fizz") } } fun CoroutineScope.buzz() = produce<String> { while (true) { // 每 1000ms 发生一个 "Buzz!" delay(1000) send("Buzz!") } } suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) { select<Unit> { // <Unit> 表示这个 select 表达式不产生任何结果值 fizz.onReceive { value -> // 这是第 1 个 select 子句 println("fizz -> '$value'") } buzz.onReceive { value -> // 这是第 2 个 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 }

这个示例程序的输出结果是:

fizz -> 'Fizz' buzz -> 'Buzz!' fizz -> 'Fizz' fizz -> 'Fizz' buzz -> 'Buzz!' fizz -> 'Fizz' buzz -> 'Fizz!'

在通道关闭时选择

如果通道已关闭, 那么 select 表达式的 onReceive 子句会失败, 并导致 select 表达式抛出一个异常. 我们可以使用 [onReceiveOrNull][onReceiveOrNull] 子句, 来对通道关闭的情况执行某个操作. 下面的示例程序还演示了 select 是一个表达式, 它会返回它的子句的结果:

suspend fun selectAorB(a: ReceiveChannel<String>, b: ReceiveChannel<String>): String = select<String> { a.onReceiveCatching { it -> val value = it.getOrNull() if (value != null) { "a -> '$value'" } else { "Channel 'a' is closed" } } b.onReceiveCatching { it -> val value = it.getOrNull() if (value != null) { "b -> '$value'" } else { "Channel 'b' is closed" } } }

假设 a 通道产生 4 次 "Hello" 字符串, b 通道产生 4 次 "World" 字符串, 我们来使用一下这个函数:

import kotlinx.coroutines.* import kotlinx.coroutines.channels.* import kotlinx.coroutines.selects.* suspend fun selectAorB(a: ReceiveChannel<String>, b: ReceiveChannel<String>): String = select<String> { a.onReceiveCatching { it -> val value = it.getOrNull() if (value != null) { "a -> '$value'" } else { "Channel 'a' is closed" } } b.onReceiveCatching { it -> val value = it.getOrNull() if (value != null) { "b -> '$value'" } else { "Channel 'b' is closed" } } } 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) { // 输出前 8 个结果 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偏向 第 1 个子句. 如果同时存在多个通道可供选择, 那么会优先选择其中的第 1 个. 在上面的示例中, 两个通道都在不断产生字符串, 因此第 1 个通道, 也就是 a, 会被优先使用. 然而, 由于我们使用了无缓冲区的通道, 因此 a 在调用 send 时有时会挂起, 因此 b 通道也有机会可以发送数据.

第 2 个现象是, 当通道被关闭时, 会立即选择 onReceiveCatching 子句.

发送时选择

选择表达式也可以使用 onSend 子句, 它可以与选择表达式的偏向性结合起来, 起到很好的作用.

下面我们来编写一个示例程序, 有一个整数值的生产者, 当主通道的消费者的消费速度跟不上生产者的发送速度时, 会把它的值改为发送到 side 通道:

fun CoroutineScope.produceNumbers(side: SendChannel<Int>) = produce<Int> { for (num in 1..10) { // 产生 10 个数值, 从 1 到 10 delay(100) // 每隔 100 ms select<Unit> { onSend(num) {} // 发送到主通道 side.onSend(num) {} // 或者发送到 side 通道 } } }

我们让消费者的运行速度变得比较慢一些, 每隔 250 ms 处理一个数值:

import kotlinx.coroutines.* import kotlinx.coroutines.channels.* import kotlinx.coroutines.selects.* fun CoroutineScope.produceNumbers(side: SendChannel<Int>) = produce<Int> { for (num in 1..10) { // 产生 10 个数值, 从 1 到 10 delay(100) // 每隔 100 ms 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 }

下面我们来看看运行结果会怎么样:

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

选择延迟的值

可以使用 onAwait 子句来选择延迟的值(Deferred value). 我们先从一个异步函数开始, 它会延迟一段随机长度的时间, 然后返回一个延迟的字符串值:

fun CoroutineScope.asyncString(time: Int) = async { delay(time.toLong()) "Waited for $time ms" }

然后我们用随机长度的延迟时间, 来启动这个函数 12 次.

fun CoroutineScope.asyncStringsList(): List<Deferred<String>> { val random = Random(3) return List(12) { asyncString(random.nextInt(1000)) } }

下面, 我们让 main 函数等待这些异步函数的第 1 个运行完毕, 然后统计仍处于激活状态的延迟值的数量. 注意, 这里我们利用了 select 表达式是 Kotlin DSL 的这种特性, 因此我们可以使用任意的代码来作为它的子句. 在这个示例程序中, 我们在一个延迟值的 List 上循环, 为每个延迟值产生一个 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

在延迟值的通道上切换

下面我们来编写一个通道生产者函数, 它从一个通道得到延迟的字符串值, 等待每一个接收到的值, 但如果下一个延迟值到达, 或者通道被关闭, 就不再等待了. 这个示例程序在同一个 select 中结合使用了 onReceiveCatching 子句 和 onAwait 子句:

fun CoroutineScope.switchMapDeferreds(input: ReceiveChannel<Deferred<String>>) = produce<String> { var current = input.receive() // 从第 1 个接收到的延迟值开始 while (isActive) { // 无限循环, 直到通道被取消/关闭 val next = select<Deferred<String>?> { // 这个 select 表达式返回下一个延迟值, 或者 null input.onReceiveCatching { update -> update.getOrNull() } current.onAwait { value -> send(value) // 如果当前正在等待的延迟值已经产生, 将它发送出去 input.receiveCatching().getOrNull() // 再继续使用从输入通道得到的下一个延迟值 } } if (next == null) { println("Channel was closed") break // 循环结束 } 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() // 从第 1 个接收到的延迟值开始 while (isActive) { // 无限循环, 直到通道被取消/关闭 val next = select<Deferred<String>?> { // 这个 select 表达式返回下一个延迟值, 或者 null input.onReceiveCatching { update -> update.getOrNull() } current.onAwait { value -> send(value) // 如果当前正在等待的延迟值已经产生, 将它发送出去 input.receiveCatching().getOrNull() // 再继续使用从输入通道得到的下一个延迟值 } } 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 }

这个示例程序的运行结果是:

BEGIN Replace END Channel was closed
最终更新: 2024/10/17