select 表达式(实验性的)

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

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

select 表达式(实验性的) - 图1

在通道中 select

我们现在有两个字符串生产者:fizzbuzz 。其中 fizz 每 500 毫秒生成一个“Fizz”字符串:

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

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

  1. fun CoroutineScope.buzz() = produce<String> {
  2. while (true) { // 每 1000 毫秒发送一个"Buzz!"
  3. delay(1000)
  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.*
  4. fun CoroutineScope.fizz() = produce<String> {
  5. while (true) { // 每 500 毫秒发送一个 "Fizz"
  6. delay(500)
  7. send("Fizz")
  8. }
  9. }
  10. fun CoroutineScope.buzz() = produce<String> {
  11. while (true) { // 每 1000 毫秒发送一个 "Buzz!"
  12. delay(1000)
  13. send("Buzz!")
  14. }
  15. }
  16. suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) {
  17. select<Unit> { // <Unit> 意味着该 select 表达式不返回任何结果
  18. fizz.onReceive { value -> // 这是第一个 select 子句
  19. println("fizz -> '$value'")
  20. }
  21. buzz.onReceive { value -> // 这是第二个 select 子句
  22. println("buzz -> '$value'")
  23. }
  24. }
  25. }
  26. fun main() = runBlocking<Unit> {
  27. //sampleStart
  28. val fizz = fizz()
  29. val buzz = buzz()
  30. repeat(7) {
  31. selectFizzBuzz(fizz, buzz)
  32. }
  33. coroutineContext.cancelChildren() // 取消 fizz 和 buzz 协程
  34. //sampleEnd
  35. }

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

select 表达式(实验性的) - 图2

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

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

通道关闭时 select

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

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

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

  1. import kotlinx.coroutines.*
  2. import kotlinx.coroutines.channels.*
  3. import kotlinx.coroutines.selects.*
  4. suspend fun selectAorB(a: ReceiveChannel<String>, b: ReceiveChannel<String>): String =
  5. select<String> {
  6. a.onReceiveCatching { it ->
  7. val value = it.getOrNull()
  8. if (value != null) {
  9. "a -> '$value'"
  10. } else {
  11. "Channel 'a' is closed"
  12. }
  13. }
  14. b.onReceiveCatching { it ->
  15. val value = it.getOrNull()
  16. if (value != null) {
  17. "b -> '$value'"
  18. } else {
  19. "Channel 'b' is closed"
  20. }
  21. }
  22. }
  23. fun main() = runBlocking<Unit> {
  24. //sampleStart
  25. val a = produce<String> {
  26. repeat(4) { send("Hello $it") }
  27. }
  28. val b = produce<String> {
  29. repeat(4) { send("World $it") }
  30. }
  31. repeat(8) { // 打印最早的八个结果
  32. println(selectAorB(a, b))
  33. }
  34. coroutineContext.cancelChildren()
  35. //sampleEnd
  36. }

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

select 表达式(实验性的) - 图3

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

  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 也有机会发送。

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

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.*
  4. fun CoroutineScope.produceNumbers(side: SendChannel<Int>) = produce<Int> {
  5. for (num in 1..10) { // 生产从 1 到 10 的 10 个数值
  6. delay(100) // 延迟 100 毫秒
  7. select<Unit> {
  8. onSend(num) {} // 发送到主通道
  9. side.onSend(num) {} // 或者发送到 side 通道
  10. }
  11. }
  12. }
  13. fun main() = runBlocking<Unit> {
  14. //sampleStart
  15. val side = Channel<Int>() // 分配 side 通道
  16. launch { // 对于 side 通道来说,这是一个很快的消费者
  17. side.consumeEach { println("Side channel has $it") }
  18. }
  19. produceNumbers(side).consumeEach {
  20. println("Consuming $it")
  21. delay(250) // 不要着急,让我们正确消化消耗被发送来的数字
  22. }
  23. println("Done consuming")
  24. coroutineContext.cancelChildren()
  25. //sampleEnd
  26. }

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

select 表达式(实验性的) - 图4

让我们看看会发生什么:

  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.*
  4. fun CoroutineScope.asyncString(time: Int) = async {
  5. delay(time.toLong())
  6. "Waited for $time ms"
  7. }
  8. fun CoroutineScope.asyncStringsList(): List<Deferred<String>> {
  9. val random = Random(3)
  10. return List(12) { asyncString(random.nextInt(1000)) }
  11. }
  12. fun main() = runBlocking<Unit> {
  13. //sampleStart
  14. val list = asyncStringsList()
  15. val result = select<String> {
  16. list.withIndex().forEach { (index, deferred) ->
  17. deferred.onAwait { answer ->
  18. "Deferred $index produced answer '$answer'"
  19. }
  20. }
  21. }
  22. println(result)
  23. val countActive = list.count { it.isActive }
  24. println("$countActive coroutines are still active")
  25. //sampleEnd
  26. }

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

select 表达式(实验性的) - 图5

该输出如下:

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

在延迟值通道上切换

我们现在来编写一个通道生产者函数,它消费一个产生延迟字符串的通道,并等待每个接收的延迟值,但它只在下一个延迟值到达或者通道关闭之前处于运行状态。此示例将 onReceiveCatchingonAwait 子句放在同一个 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.onReceiveCatching { update ->
  6. update.getOrNull()
  7. }
  8. current.onAwait { value ->
  9. send(value) // 发送当前延迟生成的值
  10. input.receiveCatching().getOrNull() // 然后使用从输入通道得到的下一个延迟值
  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.*
  4. fun CoroutineScope.switchMapDeferreds(input: ReceiveChannel<Deferred<String>>) = produce<String> {
  5. var current = input.receive() // 从第一个接收到的延迟值开始
  6. while (isActive) { // 循环直到被取消或关闭
  7. val next = select<Deferred<String>?> { // 从这个 select 中返回下一个延迟值或 null
  8. input.onReceiveCatching { update ->
  9. update.getOrNull()
  10. }
  11. current.onAwait { value ->
  12. send(value) // 发送当前延迟生成的值
  13. input.receiveCatching().getOrNull() // 然后使用从输入通道得到的下一个延迟值
  14. }
  15. }
  16. if (next == null) {
  17. println("Channel was closed")
  18. break // 跳出循环
  19. } else {
  20. current = next
  21. }
  22. }
  23. }
  24. fun CoroutineScope.asyncString(str: String, time: Long) = async {
  25. delay(time)
  26. str
  27. }
  28. fun main() = runBlocking<Unit> {
  29. //sampleStart
  30. val chan = Channel<Deferred<String>>() // 测试使用的通道
  31. launch { // 启动打印协程
  32. for (s in switchMapDeferreds(chan))
  33. println(s) // 打印每个获得的字符串
  34. }
  35. chan.send(asyncString("BEGIN", 100))
  36. delay(200) // 充足的时间来生产 "BEGIN"
  37. chan.send(asyncString("Slow", 500))
  38. delay(100) // 不充足的时间来生产 "Slow"
  39. chan.send(asyncString("Replace", 100))
  40. delay(500) // 在最后一个前给它一点时间
  41. chan.send(asyncString("END", 500))
  42. delay(1000) // 给执行一段时间
  43. chan.close() // 关闭通道……
  44. delay(500) // 然后等待一段时间来让它结束
  45. //sampleEnd
  46. }

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

select 表达式(实验性的) - 图6

这段代码的执行结果:

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