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