Kotlin 协程 - 多路复用 select()
一、概念
又叫选择表达式，是一个挂起函数，可以同时等待多个挂起结果，只取用最快恢复的那个值（即多种方式获取数据，哪个更快返回结果就用哪个）。
同时到达 select() 会优先选择先写子表达式，想随机（公平）的话使用 selectUnbiased() 替换 。
能被选择的都是 SelectClauseN 函数类型。
| public suspend inline fun <R> select(crossinline builder: SelectBuilder<R>.() -> Unit): R |
| public sealed interface SelectBuilder<in R> { public operator fun SelectClause0.invoke(block: suspend () -> R) |
| SelectClause0 | 对应事件没有返回值。例如 job.onJoin。 |
| SelectClause1 | 对应事件有返回值。例如 deffered.onAwait 和 channel.onReceive。 |
| SelectClause2 | 对应事件有返回值。此外还需要一个额外的参数，例如 Channel.onSend() 有两个参数，第一个是 Channel 数据类型的值表示即将发送的值，第二个是发送成功时的回调函数。 |
二、使用
在使用 async() 启动协程的返回类型 Deferred 中，定义了 SelectClause1 函数类型的变量 onAwait，其作用和 await() 一样，只是当其在 select() 中作为子语句时，具有“同时等待看谁最先返回”的效果。同理其它。
2.1 复用多个 job.onJoin
fun main() = runBlocking<Unit> {val job1 = launch {delay(100)println("job 1")}val job2 = launch {delay(10)println("job 2")}select {job1.onJoin { println("job 1 更快") }job2.onJoin { println("job 2 更快") }}delay(1000)
}
//打印：
//job 2
//job 2 更快
//job 12.2 复用多个 deffered.onAwait
| public interface Deferred<out T> : Job { public val onAwait: SelectClause1<T> //等效await() ｝ |
fun main() = runBlocking {val defferedCache = async {delay(10)"Cache"}val defferedLocal = async {delay(100)"Local"}val defferedRemote = async {delay(1000)"Remote"}val result = select {defferedCache.onAwait { println("最快的是$it") }defferedLocal.onAwait { println("最快的是$it") }defferedRemote.onAwait { println("最快的是$it") }}delay(2000)println(result) //打印：最快的是Cache
}2.3 复用多个 channel.onReceive
| public interface SendChannel<in E> { public val onSend: SelectClause2<E, SendChannel<E>> //等效send() } |
| public interface ReceiveChannel<out E> { public val onReceive: SelectClause1<E> //等效receive() public suspend fun receiveCatching(): ChannelResult<E> //等效receiveCatching() } //select() 中的 onReceive() 在已经关闭的通道执行会发生失败，并导致相应的 select() 抛出异常，使用 onReceiveCatching() 在关闭通道时执行特定操作。 |
suspend fun getDataFromLocal() = withContext(Dispatchers.IO) { "Local" }
suspend fun getDataFromRemote() = withContext(Dispatchers.IO) { "Remote" }@OptIn(ExperimentalCoroutinesApi::class)
fun main() = runBlocking {val produceLocal = produce { send(getDataFromLocal()) }val produceRemote = produce { send(getDataFromRemote()) }val result = select {produceLocal.onReceive { it }produceRemote.onReceive { it }}
// val result = select {
// produceLocal.onReceiveCatching { it.getOrNull() ?: "Channel已关闭：produceLocal" }
// produceRemote.onReceiveCatching { it.getOrNull() ?: "Channel已关闭：produceRemote " }
// }println("结果更快的是：$result")
}