目录

select 表达式（实验性的）

select 表达式可以同时等待多个挂起函数，并 选择 第一个可用的。

Select 表达式在 kotlinx.coroutines 中是一个实验性的特性。这些 API 在 kotlinx.coroutines 库即将到来的更新中可能会发生改变。

在通道中 select

我们现在有两个字符串生产者：fizz 和 buzz 。其中 fizz 每 300 毫秒生成一个“Fizz”字符串：

1. fun CoroutineScope.fizz() = produce<String> {
2. while (true) { // 每 300 毫秒发送一个 "Fizz"
3. delay(300)
4. send("Fizz")
5. }
6. }

接着 buzz 每 500 毫秒生成一个 “Buzz!” 字符串：

1. fun CoroutineScope.buzz() = produce<String> {
2. while (true) { // 每 500 毫秒发送一个"Buzz!"
3. delay(500)
4. send("Buzz!")
5. }
6. }

使用 receive 挂起函数，我们可以从两个通道接收 其中一个 的数据。 但是 select 表达式允许我们使用其 onReceive 子句 同时 从两者接收：

1. suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) {
2. select<Unit> { // <Unit> 意味着该 select 表达式不返回任何结果
3. fizz.onReceive { value ->  // 这是第一个 select 子句
4. println("fizz -> '$value'")
5. }
6. buzz.onReceive { value ->  // 这是第二个 select 子句
7. println("buzz -> '$value'")
8. }
9. }
10. }

让我们运行代码 7 次：

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.channels.*
3. import kotlinx.coroutines.selects.*

5. fun CoroutineScope.fizz() = produce<String> {
6. while (true) { // 每 300 毫秒发送一个 "Fizz"
7. delay(300)
8. send("Fizz")
9. }
10. }

12. fun CoroutineScope.buzz() = produce<String> {
13. while (true) { // 每 500 毫秒发送一个 "Buzz!"
14. delay(500)
15. send("Buzz!")
16. }
17. }

19. suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) {
20. select<Unit> { // <Unit> 意味着该 select 表达式不返回任何结果
21. fizz.onReceive { value ->  // 这是第一个 select 子句
22. println("fizz -> '$value'")
23. }
24. buzz.onReceive { value ->  // 这是第二个 select 子句
25. println("buzz -> '$value'")
26. }
27. }
28. }

30. fun main() = runBlocking<Unit> {
31. //sampleStart
32. val fizz = fizz()
33. val buzz = buzz()
34. repeat(7) {
35. selectFizzBuzz(fizz, buzz)
36. }
37. coroutineContext.cancelChildren() // 取消 fizz 和 buzz 协程
38. //sampleEnd
39. }

可以在这里获取完整代码。

这段代码的执行结果如下：

1. fizz -> 'Fizz'
2. buzz -> 'Buzz!'
3. fizz -> 'Fizz'
4. fizz -> 'Fizz'
5. buzz -> 'Buzz!'
6. fizz -> 'Fizz'
7. buzz -> 'Buzz!'

通道关闭时 select

select 中的 onReceive 子句在已经关闭的通道执行会发生失败，并导致相应的 select 抛出异常。我们可以使用 onReceiveOrNull 子句在关闭通道时执行特定操作。以下示例还显示了 select 是一个返回其查询方法结果的表达式：

1. suspend fun selectAorB(a: ReceiveChannel<String>, b: ReceiveChannel<String>): String =
2. select<String> {
3. a.onReceiveOrNull { value ->
4. if (value == null)
5. "Channel 'a' is closed"
6. else
7. "a -> '$value'"
8. }
9. b.onReceiveOrNull { value ->
10. if (value == null)
11. "Channel 'b' is closed"
12. else
13. "b -> '$value'"
14. }
15. }

注意，onReceiveOrNull 是一个仅在用于不可空元素的通道上定义的扩展函数，以使关闭的通道与空值之间不会出现意外的混乱。

现在有一个生成四次“Hello”字符串的 a 通道， 和一个生成四次“World”字符串的 b 通道，我们在这两个通道上使用它：

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.channels.*
3. import kotlinx.coroutines.selects.*

5. suspend fun selectAorB(a: ReceiveChannel<String>, b: ReceiveChannel<String>): String =
6. select<String> {
7. a.onReceiveOrNull { value ->
8. if (value == null)
9. "Channel 'a' is closed"
10. else
11. "a -> '$value'"
12. }
13. b.onReceiveOrNull { value ->
14. if (value == null)
15. "Channel 'b' is closed"
16. else
17. "b -> '$value'"
18. }
19. }

21. fun main() = runBlocking<Unit> {
22. //sampleStart
23. val a = produce<String> {
24. repeat(4) { send("Hello $it") }
25. }
26. val b = produce<String> {
27. repeat(4) { send("World $it") }
28. }
29. repeat(8) { // 打印最早的八个结果
30. println(selectAorB(a, b))
31. }
32. coroutineContext.cancelChildren()
33. //sampleEnd
34. }

可以在这里获取完整代码。

这段代码的结果非常有趣，所以我们会在更多细节中分析它：

1. a -> 'Hello 0'
2. a -> 'Hello 1'
3. b -> 'World 0'
4. a -> 'Hello 2'
5. a -> 'Hello 3'
6. b -> 'World 1'
7. Channel 'a' is closed
8. Channel 'a' is closed

有几个结果可以通过观察得出。

首先，select 偏向于 第一个子句，当可以同时选到多个子句时， 第一个子句将被选中。在这里，两个通道都在不断地生成字符串，因此 a 通道作为 select 中的第一个子句获胜。然而因为我们使用的是无缓冲通道，所以 a 在其调用 send 时会不时地被挂起，进而 b 也有机会发送。

第二个观察结果是，当通道已经关闭时， 会立即选择 onReceiveOrNull。

Select 以发送

Select 表达式具有 onSend 子句，可以很好的与选择的偏向特性结合使用。

我们来编写一个整数生成器的示例，当主通道上的消费者无法跟上它时，它会将值发送到 side 通道上：

1. fun CoroutineScope.produceNumbers(side: SendChannel<Int>) = produce<Int> {
2. for (num in 1..10) { // 生产从 1 到 10 的 10 个数值
3. delay(100) // 延迟 100 毫秒
4. select<Unit> {
5. onSend(num) {} // 发送到主通道
6. side.onSend(num) {} // 或者发送到 side 通道
7. }
8. }
9. }

消费者将会非常缓慢，每个数值处理需要 250 毫秒：

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.channels.*
3. import kotlinx.coroutines.selects.*

5. fun CoroutineScope.produceNumbers(side: SendChannel<Int>) = produce<Int> {
6. for (num in 1..10) { // 生产从 1 到 10 的 10 个数值
7. delay(100) // 延迟 100 毫秒
8. select<Unit> {
9. onSend(num) {} // 发送到主通道
10. side.onSend(num) {} // 或者发送到 side 通道
11. }
12. }
13. }

15. fun main() = runBlocking<Unit> {
16. //sampleStart
17. val side = Channel<Int>() // 分配 side 通道
18. launch { // 对于 side 通道来说，这是一个很快的消费者
19. side.consumeEach { println("Side channel has $it") }
20. }
21. produceNumbers(side).consumeEach {
22. println("Consuming $it")
23. delay(250) // 不要着急，让我们正确消化消耗被发送来的数字
24. }
25. println("Done consuming")
26. coroutineContext.cancelChildren()
27. //sampleEnd
28. }

可以在这里获取完整代码。

让我们看看会发生什么：

1. Consuming 1
2. Side channel has 2
3. Side channel has 3
4. Consuming 4
5. Side channel has 5
6. Side channel has 6
7. Consuming 7
8. Side channel has 8
9. Side channel has 9
10. Consuming 10
11. Done consuming

Select 延迟值

延迟值可以使用 onAwait 子句查询。 让我们启动一个异步函数，它在随机的延迟后会延迟返回字符串：

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

让我们随机启动十余个异步函数，每个都延迟随机的时间。

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

现在 main 函数在等待第一个函数完成，并统计仍处于激活状态的延迟值的数量。注意，我们在这里使用 select 表达式事实上是作为一种 Kotlin DSL， 所以我们可以用任意代码为它提供子句。在这种情况下，我们遍历一个延迟值的队列，并为每个延迟值提供 onAwait 子句的调用。

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.selects.*
3. import java.util.*

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

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

15. fun main() = runBlocking<Unit> {
16. //sampleStart
17. val list = asyncStringsList()
18. val result = select<String> {
19. list.withIndex().forEach { (index, deferred) ->
20. deferred.onAwait { answer ->
21. "Deferred $index produced answer '$answer'"
22. }
23. }
24. }
25. println(result)
26. val countActive = list.count { it.isActive }
27. println("$countActive coroutines are still active")
28. //sampleEnd
29. }

可以在这里获取完整代码。

该输出如下：

1. Deferred 4 produced answer 'Waited for 128 ms'
2. 11 coroutines are still active

在延迟值通道上切换

我们现在来编写一个通道生产者函数，它消费一个产生延迟字符串的通道，并等待每个接收的延迟值，但它只在下一个延迟值到达或者通道关闭之前处于运行状态。此示例将 onReceiveOrNull 和 onAwait 子句放在同一个 select 中：

1. fun CoroutineScope.switchMapDeferreds(input: ReceiveChannel<Deferred<String>>) = produce<String> {
2. var current = input.receive() // 从第一个接收到的延迟值开始
3. while (isActive) { // 循环直到被取消或关闭
4. val next = select<Deferred<String>?> { // 从这个 select 中返回下一个延迟值或 null
5. input.onReceiveOrNull { update ->
6. update // 替换下一个要等待的值
7. }
8. current.onAwait { value ->
9. send(value) // 发送当前延迟生成的值
10. input.receiveOrNull() // 然后使用从输入通道得到的下一个延迟值
11. }
12. }
13. if (next == null) {
14. println("Channel was closed")
15. break // 跳出循环
16. } else {
17. current = next
18. }
19. }
20. }

为了测试它，我们将用一个简单的异步函数，它在特定的延迟后返回特定的字符串：

1. fun CoroutineScope.asyncString(str: String, time: Long) = async {
2. delay(time)
3. str
4. }

main 函数只是启动一个协程来打印 switchMapDeferreds 的结果并向它发送一些测试数据：

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.channels.*
3. import kotlinx.coroutines.selects.*

5. fun CoroutineScope.switchMapDeferreds(input: ReceiveChannel<Deferred<String>>) = produce<String> {
6. var current = input.receive() // 从第一个接收到的延迟值开始
7. while (isActive) { // 循环直到被取消或关闭
8. val next = select<Deferred<String>?> { // 从这个 select 中返回下一个延迟值或 null
9. input.onReceiveOrNull { update ->
10. update // 替换下一个要等待的值
11. }
12. current.onAwait { value ->
13. send(value) // 发送当前延迟生成的值
14. input.receiveOrNull() // 然后使用从输入通道得到的下一个延迟值
15. }
16. }
17. if (next == null) {
18. println("Channel was closed")
19. break // 跳出循环
20. } else {
21. current = next
22. }
23. }
24. }

26. fun CoroutineScope.asyncString(str: String, time: Long) = async {
27. delay(time)
28. str
29. }

31. fun main() = runBlocking<Unit> {
32. //sampleStart
33. val chan = Channel<Deferred<String>>() // 测试使用的通道
34. launch { // 启动打印协程
35. for (s in switchMapDeferreds(chan))
36. println(s) // 打印每个获得的字符串
37. }
38. chan.send(asyncString("BEGIN", 100))
39. delay(200) // 充足的时间来生产 "BEGIN"
40. chan.send(asyncString("Slow", 500))
41. delay(100) // 不充足的时间来生产 "Slow"
42. chan.send(asyncString("Replace", 100))
43. delay(500) // 在最后一个前给它一点时间
44. chan.send(asyncString("END", 500))
45. delay(1000) // 给执行一段时间
46. chan.close() // 关闭通道……
47. delay(500) // 然后等待一段时间来让它结束
48. //sampleEnd
49. }

可以在这里获取完整代码。

这段代码的执行结果：

1. BEGIN
2. Replace
3. END
4. Channel was closed