【Kotlin】select简介

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

打印如下。