UP | HOME

Select表达式

Table of Contents

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

在通道中 select

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

fun CoroutineScope.fizz() = produce {
    while (true) {
        // 每 300 毫秒发送一个 "Fizz"
        delay(300)
        send("Fizz")
    }
}

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

fun CoroutineScope.buzz() = produce {
    while (true) {
        // 每 300 毫秒发送一个 "Fizz"
        delay(500)
        send("Buzz")
    }
}

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

suspend fun selectFizzBuzz(fizz: ReceiveChannel, buzz: ReceiveChannel) {
    select { // 意味着该 select 表达式不返回任何结果
             fizz.onReceive {
                 value -> // 这是第一个 select 子句
                 println("fizz -> '$value'")
             }
             buzz.onReceive { value -> // 这是第二个 select 子句
                              println("buzz -> '$value'")
             }
    }
}

运行代码 7 次： 

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.cancelChildren
import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.channels.produce
import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.selects.select

fun CoroutineScope.fizz() = produce {
    while (true) {
        // 每 300 毫秒发送一个 "Fizz"
        delay(300)
        send("Fizz")
    }
}

fun CoroutineScope.buzz() = produce {
    while (true) {
        // 每 300 毫秒发送一个 "Fizz"
        delay(500)
        send("Buzz")
    }
}

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

fun main() = runBlocking {
    val fizz = fizz()
    val buzz = buzz()
    repeat(7) {
        selectFizzBuzz(fizz, buzz)
    }
    coroutineContext.cancelChildren() // 取消 fizz 和 buzz 协程
}

运行的结果如下： 

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

通道关闭时 select

select 中的 onReceive 子句在已经关闭的通道执行会发生失败，并导致相应的 select 抛出异常

可以使用 onReceiveCatching 子句在关闭通道时执行特定操作。以下示例还显示了 select 是一个返回其查询方法结果的表达式： 

suspend fun selectAorB(a: ReceiveChannel<String>, b: ReceiveChannel<String>): String =
select {
    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"
                          }
    }
}

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

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

import kotlinx.coroutines.cancelChildren
import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.channels.produce
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.selects.select

suspend fun selectAorB(a: ReceiveChannel<String>, b: ReceiveChannel<String>): String =
    select {
        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 {
    val a = produce {
        repeat(4) {
            send("Hello $it")
        }
    }
    val b = produce {
        repeat(4) {
            send("World $it")
        }
    }
    repeat(8) { // 打印最早的八个结果
        println(selectAorB(a, b))
    }
    coroutineContext.cancelChildren()
}

这段代码的结果非常有趣 

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 偏向于 第一个子句，当可以同时选到多个子句时， 第一个子句将被选中 

          在这里，两个通道都在不断地生成字符串，因此 a 通道作为 select 中的第一个子句获胜

      然而因为使用的是无缓冲通道，所以 a 在其调用 send 时会不时地被挂起，进而 b 也有机会发送
  • 当通道已经关闭时， 会立即选择 onReceiveCatching

select 发送通道

Select 表达式具有 onSend 子句，可以有选择的发送到某些通道。下面是一个整数生成器的示例，当主通道上的消费者无法跟上它时，它会将值发送到 side 通道上： 

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

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

fun main() = runBlocking {
    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()
}

看看会发生什么： 

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

选择延迟值

Select 延迟值可以使用 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 函数在等待第一个函数完成，并统计仍处于激活状态的延迟值的数量： 

fun main() = runBlocking {
    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")
}

输出并不固定，类似于： 

Deferred 6 produced answer 'Waited for 43 ms'
11 coroutines are still active

    注意，在这里使用 select 表达式事实上是作为一种 Kotlin DSL， 因此可以用任意代码为它提供子句

    在这种情况下，遍历一个延迟值的队列，并为每个延迟值提供 onAwait 子句的调用

在延迟通道上切换

现在来编写一个通道生产者函数，它消费一个产生延迟字符串的通道，并等待每个接收的延迟值，但它只在下一个延迟值到达或者通道关闭之前处于运行状态。此示例将 onReceiveCatching 和 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.onReceiveCatching { update ->
                update.getOrNull()
            }
            current.onAwait { value ->
                send(value) // send value that current deferred has produced
                input.receiveCatching().getOrNull() // 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 的结果并向它发送一些测试数据： 

fun main() = runBlocking {
    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
}

这段代码的执行结果： 

BEGIN
Replace
END
Channel was closed
Previous：共享状态 Home：协程