kotlin协程——>select 表达式（实验性的）

select 表达式（实验性的）

　　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) { // 每 500 毫秒发送?个"Buzz!"
        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 次：

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'
buzz -> 'Buzz!'

通道关闭时 select

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

suspend fun selectAorB(a: ReceiveChannel, b: ReceiveChannel): String =
    select {
        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 通道，我们在这两个通道上使?它：

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

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

Select 以发送

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

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

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

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

val side = Channel() // 分配 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> {
    val random = Random(3)
    return List(12) { asyncString(random.nextInt(1000)) }
}

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

val list = asyncStringsList()
val result = select {
    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 4 produced answer 'Waited for 128 ms'
11 coroutines are still active

在延迟值通道上切换

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

fun CoroutineScope.switchMapDeferreds(input: ReceiveChannel>) =
    produce {
        var current = input.receive() // 从第?个接收到的延迟值开始
        while (isActive) { // 循环直到被取消或关闭
            val next = select?> { // 从这个 select 中返回下?个延迟值或 null
                input.onReceiveOrNull { update ->
                    update // 替换下?个要等待的值
                }
                current.onAwait { value ->
                    send(value) // 发送当前延迟?成的值
                    input.receiveOrNull() // 然后使?从输?通道得到的下?个延迟值
                }
            }
            if (next == null) {
                println("Channel was closed")
                break // 跳出循环
            } else {
                current = next
            }
        }
    }

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

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

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

val chan = Channel>() // 测试使?的通道
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) // 然后等待?段时间来让它结束

　　这段代码的执?结果：

BEGIN
Replace
END
Channel was closed

kotlin基础

kotlin协程——>select 表达式（实验性的）

相关

标签