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)
        public operator fun <Q> SelectClause1<Q>.invoke(block: suspend (Q) -> R)
        public operator fun <P, Q> SelectClause2<P, Q>.invoke(param: P, block: suspend (Q) -> R)
        public operator fun <P, Q> SelectClause2<P?, Q>.invoke(block: suspend (Q) -> R): Unit = invoke(null, block)
}

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 1

2.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")
}