1 前言
协程的 select 是一种用于异步操作的选择器,它允许同时等待多个挂起函数的结果,并在其中一个完成时执行相应的操作。
能够被 select 的事件都是 SelectClause,在 select.kt 中有定义,如下。
public 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)
...
}
public interface SelectClause0 {
public fun <R> registerSelectClause0(select: SelectInstance<R>, block: suspend () -> R)
}
public interface SelectClause1<out Q> {
public fun <R> registerSelectClause1(select: SelectInstance<R>, block: suspend (Q) -> R)
}
public interface SelectClause2<in P, out Q> {
public fun <R> registerSelectClause2(select: SelectInstance<R>, param: P, block: suspend (Q) -> R)
}
internal class SelectBuilderImpl<in R>(
private val uCont: Continuation<R>
) : LockFreeLinkedListHead(), SelectBuilder<R>,
SelectInstance<R>, Continuation<R>, CoroutineStackFrame
{
override fun SelectClause0.invoke(block: suspend () -> R) {
registerSelectClause0(this@SelectBuilderImpl, block)
}
override fun <Q> SelectClause1<Q>.invoke(block: suspend (Q) -> R) {
registerSelectClause1(this@SelectBuilderImpl, block)
}
override fun <P, Q> SelectClause2<P, Q>.invoke(param: P, block: suspend (Q) -> R) {
registerSelectClause2(this@SelectBuilderImpl, param, block)
}
}
2 select 在 Job 中的应用
1)应用
fun main() {
CoroutineScope(Dispatchers.Default).launch {
var job1 = launchJob("job-1", 100)
var job2 = launchJob("job-2", 200)
var res = select {
job1.onJoin { "select: 1" }
job2.onJoin { "select: 2" }
}
println(res) // 打印: select: 1
}
Thread.sleep(1000) // 阻塞当前线程, 避免程序过早结束, 协程提前取消
}
suspend fun launchJob(tag: String, delayTime: Long): Job =
CoroutineScope(Dispatchers.Default).launch {
println("tag")
delay(delayTime)
}
打印如下。
tag
tag
select: 1
2)onJoin 源码
onJoin 是 Job 中定义的属性。
public val onJoin: SelectClause0
说明:在调用 job1.onJoin { xxx } 时,等价于调用了 SelectClause0.invoke 函数,也等价于调用了 SelectClause0.registerSelectClause0 函数。
3 select 在 Deferred 中的应用
1)应用
fun main() {
CoroutineScope(Dispatchers.Default).launch {
var task1 = asyncTask("task-1", 100)
var task2 = asyncTask("task-2", 200)
var res = select {
task1.onAwait { "select: $it" }
task2.onAwait { "select: $it" }
}
println(res) // 打印: select: task-1
}
Thread.sleep(1000) // 阻塞当前线程, 避免程序过早结束, 协程提前取消
}
suspend fun asyncTask(tag: String, delayTime: Long): Deferred<String> =
CoroutineScope(Dispatchers.Default).async {
delay(delayTime)
tag
}
打印如下。
select: task-1
2)onAwait 源码
onAwait 是 Deferred 中定义的属性。
public val onAwait: SelectClause1<T>
说明:在调用 task1.onAwait { xxx } 时,等价于调用了 SelectClause1.invoke 函数,也等价于调用了 SelectClause1.registerSelectClause1 函数。
4 select 在 Channel 中的应用
4.1 onSend
1)应用
fun main() {
var channels = List(2) { Channel<String>() }
CoroutineScope(Dispatchers.Default).launch {
var res = select {
channels[0].onSend("select-1") { "task-1" }
channels[1].onSend("select-2") { "task-2" }
}
println("res=$res") // 打印: res=task-2
}
receiveTask(200, channels[0])
receiveTask(100, channels[1])
Thread.sleep(1000) // 阻塞当前线程, 避免程序过早结束, 协程提前取消
}
fun receiveTask(delayTime: Long, channel: Channel<String>) {
CoroutineScope(Dispatchers.Default).launch {
delay(delayTime)
var element = channel.receive()
println("receive: $element")
}
}
打印如下。
receive: select-2
res=task-2
2)onSend 源码
onSend 是 Channel 中定义的属性。
public val onSend: SelectClause1<E>
说明:在调用 channels[0].onSend(xxx) { yyy } 时,等价于调用了 SelectClause2.invoke 函数,也等价于调用了 SelectClause2.registerSelectClause2 函数。
4.2 onReceive
1)应用
fun main() {
var channels = List(2) { Channel<String>() }
sendTask("task-1", 200, channels[0])
sendTask("task-2", 100, channels[1])
CoroutineScope(Dispatchers.Default).launch {
var res = select {
channels[0].onReceive { "select: $it" }
channels[1].onReceive { "select: $it" }
}
println(res) // 打印: select: task-2
}
Thread.sleep(1000) // 阻塞当前线程, 避免程序过早结束, 协程提前取消
}
fun sendTask(tag: String, delayTime: Long, channel: Channel<String>) {
CoroutineScope(Dispatchers.Default).launch {
delay(delayTime)
channel.send(tag)
}
}
打印如下。
select: task-2
2)onReceive 源码
onReceive 是 Channel 中定义的属性。
public val onReceive: SelectClause1<E>
说明:在调用 channels[0].onReceive { xxx } 时,等价于调用了 SelectClause1.invoke 函数,也等价于调用了 SelectClause1.registerSelectClause1 函数。