注册

Kotlin 协程 Select:看我如何多路复用

前言

协程通信三剑客:Channel、Select、Flow,上篇已经分析了Channel的深水区,本篇将会重点分析Select的使用及原理。
通过本篇文章,你将了解到:

  1. Select 的引入
  2. Select 的使用
  3. Invoke函数 的妙用
  4. Select 的原理
  5. Select 注意事项

1. Select 的引入

多路数据的选择

串行执行

如今的二维码识别应用场景越来越广了,早期应用比较广泛的识别SDK如zxing、zbar,它们各有各的特点,也存在识别不出来的情况,为了将两者优势结合起来,我们想到的方法是同一份二维码图片分别给两者进行识别。
如下:

    //从zxing 获取二维码信息
suspend fun getQrcodeInfoFromZxing(bitmap: Bitmap?): String {
//模拟耗时
delay(2000)
return "I'm fish"
}

//从zbar 获取二维码信息
suspend fun getQrcodeInfoFromZbar(bitmap: Bitmap?): String {
delay(1000)
return "I'm fish"
}

fun testSelect() {
runBlocking {
var bitmap = null
var starTime = System.currentTimeMillis()
var qrcoe1 = getQrcodeInfoFromZxing(bitmap)
var qrcode2 = getQrcodeInfoFromZbar(bitmap)
println("qrcode1=$qrcoe1 qrcode2=$qrcode2 useTime:${System.currentTimeMillis() - starTime} ms")
}
}

查看打印,最后花费的时间:

qrcode1=I'm fish qrcode2=I'm fish useTime:3013 ms

当然这是串行的方式效率比较低,我们想到了用协程来优化它。

协程并行执行

如下:

    fun testSelect1() {
var bitmap = null;
var starTime = System.currentTimeMillis()
var deferredZxing = GlobalScope.async {
getQrcodeInfoFromZxing(bitmap)
}

var deferredZbar = GlobalScope.async {
getQrcodeInfoFromZbar(bitmap)
}

runBlocking {
//挂起等待识别结果
var qrcoe1 = deferredZxing.await()
//挂起等待识别结果
var qrcode2 = deferredZbar.await()
println("qrcode1=$qrcoe1 qrcode2=$qrcode2 useTime:${System.currentTimeMillis() - starTime} ms")
}
}

查看打印,最后花费的时间:

qrcode1=I'm fish qrcode2=I'm fish useTime:2084 ms

可以看出,花费时间明显变少了。
与上个Demo 相比,虽然识别过程是放在协程里并行执行的,但是在等待识别结果却是串行的。我们引入两个识别库的初衷是哪个识别快就用哪个的结果,为了达成这个目的,传统的方式是:

同时监听并记录识别结果的返回。

同时监听多路结果

如下:

    fun testSelect2() {
var bitmap = null;
var starTime = System.currentTimeMillis()
var deferredZxing = GlobalScope.async {
getQrcodeInfoFromZxing(bitmap)
}

var deferredZbar = GlobalScope.async {
getQrcodeInfoFromZbar(bitmap)
}

var isEnd = false
var result: String? = null
GlobalScope.launch {
if (!isEnd) {
//没有结束,则继续识别
var resultTmp = deferredZxing.await()
if (!isEnd) {
//识别没有结束,说明自己是第一个返回结果的
result = resultTmp
println("zxing recognize ok useTime:${System.currentTimeMillis() - starTime} ms")
//标记识别结束
isEnd = true
}
}
}

GlobalScope.launch {
if (!isEnd) {
var resultTmp = deferredZbar.await()
if (!isEnd) {
//识别没有结束,说明自己是第一个返回结果的
result = resultTmp
println("zbar recognize ok useTime:${System.currentTimeMillis() - starTime} ms")
isEnd = true
}
}
}

//检测是否有结果返回
runBlocking {
while (!isEnd) {
delay(1)
}
println("recognize result:$result")
}
}

通过检测isEnd 标记来判断是否有某个模块返回结果。
结果如下:

  • zbar recognize ok useTime:1070 ms
  • recognize result:I'm fish

由于模拟设定的zbar 解析速度快,因此每次都是采纳的是zbar的结果,所花费的时间大幅减少了,该结果符合预期。

Select 闪亮登场

虽说上个Demo结果符合预期,但是多了很多额外的代码、多引入了其它协程,并且需要子模块对标记进行赋值(对"isEnd"进行赋值),没有达到解耦的目的。我们希望子模块的任务是单一且闭环的,如果能在一个函数里统一检测结果的返回就好了。
Select 就是为了解决多路数据的选择而生的。
来看看它是怎么解决该问题的:

    fun testSelect3() {
var bitmap = null;
var starTime = System.currentTimeMillis()
var deferredZxing = GlobalScope.async {
getQrcodeInfoFromZxing(bitmap)
}
var deferredZbar = GlobalScope.async {
getQrcodeInfoFromZbar(bitmap)
}
runBlocking {
//通过select 监听zxing、zbar 结果返回
var result = select<String> {
//监听zxing
deferredZxing.onAwait {value->
//value 为deferredZxing 识别的结果
"zxing result $value"
}

//监听zbar
deferredZbar.onAwait { value->
"zbar result $value"
}
}

//运行到此,说明已经有结果返回
println("result from $result useTime:${System.currentTimeMillis() - starTime}")
}
}

结果如下:

result from zbar result I'm fish useTime:1079

符合预期,同时可以看出:相比上个Demo,这样写简洁了许多。

2. Select 的使用

除了可以监听async的结果,Select 还可以监听Channel的发送方/接收方 数据,我们以监听接收方数据为例:

    fun testSelect4() {
runBlocking {
var bitmap = null;
var starTime = System.currentTimeMillis()
var receiveChannelZxing = produce {
//生产数据
var result = getQrcodeInfoFromZxing(bitmap)
//发送数据
send(result)
}

var receiveChannelZbar = produce {
var result = getQrcodeInfoFromZbar(bitmap)
send(result)
}

var result = select<String> {
//监听是否有数据发送过来
receiveChannelZxing.onReceive {
value->"zxing result $value"
}

receiveChannelZbar.onReceive {
value->"zbar result $value"
}
}

println("result from $result useTime:${System.currentTimeMillis() - starTime}")
}
}

结果如下:

result from zbar result I'm fish useTime:1028

不论是async还是Channel,Select 都可以监听它们的数据,从而形成多路复用的效果。

image.png

在监听协程里调用select 表达式,表达式{}内声明需要监听的协程的数据,对于select 来说有两种场景:

  1. 没有数据,则select 挂起协程并等待直到其它协程数据准备完成后再次恢复select 所在的协程。
  2. 有数据,则select 正常执行并返回获取的数据。

3. Invoke函数 的妙用

在分析Select 原理之前,需要弄明白invoke函数的原理。
对于Kotlin 类来说,都可以重写其invoke函数。

    operator fun invoke():String {
return "I'm fish"
}

如上,重写了SelectDemo里的invoke函数,和普通成员函数一样,我们可以通过对象调用它。

fun main(args: Array<String>) {
var selectDemo = SelectDemo()
var result = selectDemo.invoke()
println("result:$result")
}

当然,可以进一步简化:

fun main(args: Array<String>) {
var selectDemo = SelectDemo()
var result = selectDemo()
println("result:$result")
}

这里涉及到了kotlin的语法糖:对象居然可以像函数一样调用。
作为函数,invoke 当然也可以接收高阶函数作为参数:

    operator fun invoke(block: (Int) -> String): String {
return block(3)
}

fun main(args: Array<String>) {
var selectDemo = SelectDemo()
var result = selectDemo { age ->
when (age) {
3 -> "I'm fish3"
4 -> "I'm fish4"
else -> "error"
}
}
println("result:$result")
}

因此,当看到对象作为函数调用时,实际上调用的是invoke函数,具体的逻辑需要查看其invoke函数的实现。

4. Select 的原理

上篇分析过Channel,因此本篇趁热打铁,通过Select 监听Channel数据的变化来分析其原理,为方便讲解,我们先以监听一个Channel的为例。
先从select 表达式本身入手。

    fun testSelect5() {
runBlocking {
var starTime = System.currentTimeMillis()
var receiveChannelZxing = produce {
//发送数据
send("I'm fish")
}

//确保channel 数据已经send
delay(1000)
var result = select<String> {
//监听是否有数据发送过来
receiveChannelZxing.onReceive { value ->
"zxing result $value"
}
}
println("result from $result useTime:${System.currentTimeMillis() - starTime}")
}
}

select 是挂起函数,因此协程运行到此有可能被挂起。

#Select.kt
public suspend inline fun <R> select(crossinline builder: SelectBuilder<R>.() -> Unit): R {
//...
return suspendCoroutineUninterceptedOrReturn { uCont ->
//传入父协程体
val scope = SelectBuilderImpl(uCont)
try {
//执行builder
builder(scope)
} catch (e: Throwable) {
scope.handleBuilderException(e)
}
//通过返回值判断是否需要挂起协程
scope.getResult()
}
}

重点看builder(scope),builder 是高阶函数,实际上就是执行了select花括号里的内容,而它里面就是监听数据是否返回。

receiveChannelZxing.onReceive
刚开始看的时候势必以为onReceive是个函数,然而它是ReceiveChannel 里的成员变量:

#Channel.kt
public val onReceive: SelectClause1<E>

通过上一节的分析可知,关键是要找到SelectClause1 的invoke的实现。

#Select.kt
public interface SelectBuilder<in R> {
//block 有个入参
//声明了SelectClause1的扩展函数invoke
public operator fun <Q> SelectClause1<Q>.invoke(block: suspend (Q) -> R)
}

override fun <Q> SelectClause1<Q>.invoke(block: suspend (Q) -> R) {
//SelectBuilderImpl 实现了 SelectClause1 的invoke函数
registerSelectClause1(this@SelectBuilderImpl, block)
}

再看onReceive 的赋值:

#AbstractChannel.kt
final override val onReceive: SelectClause1<E>
get() = object : SelectClause1<E> {
@Suppress("UNCHECKED_CAST")
override fun <R> registerSelectClause1(select: SelectInstance<R>, block: suspend (E) -> R) {
registerSelectReceiveMode(select, RECEIVE_THROWS_ON_CLOSE, block as suspend (Any?) -> R)
}
}

因此,简单总结调用栈如下:

当调用receiveChannelZxing.onReceive{},实际上调用了SelectClause1.invoke(),而它里面又调用了SelectClause1.registerSelectClause1(),最终调用了AbstractChannel.registerSelectReceiveMode。

AbstractChannel. registerSelectReceiveMode

#AbstractChannel.kt
private fun <R> registerSelectReceiveMode(select: SelectInstance<R>, receiveMode: Int, block: suspend (Any?) -> R) {
while (true) {
//如果已经有结果了,则直接返回------->①
if (select.isSelected) return
if (isEmptyImpl) {
//没有发送者在等待,则入队等待,并返回 ------->②
if (enqueueReceiveSelect(select, block, receiveMode)) return
} else {
//直接取出值------->③
val pollResult = pollSelectInternal(select)
when {
pollResult === ALREADY_SELECTED -> return
pollResult === POLL_FAILED -> {} // retry
pollResult === RETRY_ATOMIC -> {} // retry
//调用block------->④
else -> block.tryStartBlockUnintercepted(select, receiveMode, pollResult)
}
}
}
}

分为4个点,接着来一一分析。

select 同时监听多个值,若是有1个符合要求的数据返回了,那么该isSelected 标记为true,当检测到该标记为true时直接退出。
结合之前的Demo,zbar 已经识别出结果了,当select 检测zxing的结果时直接返回。

#AbstractChannel.kt
private fun <R> enqueueReceiveSelect(
select: SelectInstance<R>,
block: suspend (Any?) -> R,
receiveMode: Int
): Boolean {
//构造为Node元素
val node = AbstractChannel.ReceiveSelect(this, select, block, receiveMode)
//添加到Channel队列里
val result = enqueueReceive(node)
if (result) select.disposeOnSelect(node)
return result

当select 时,发现Channel里没有数据,说明Channel还没有开始send,因此构造了Node(ReceiveSelect)加入到Channel queue里。当send数据时,会查找queue里是否有接收者等待,若有则调用Node(ReceiveSelect.completeResumeReceive):

#AbstractChannel.kt
override fun completeResumeReceive(value: E) {
block.startCoroutineCancellable(
if (receiveMode == RECEIVE_RESULT) ChannelResult.success(value) else value,
select.completion,
resumeOnCancellationFun(value)
)
}

block 被调度执行,最后会恢复select 协程的执行。


取出数据,并尝试恢复send协程。


在③的基础上,拿到数据后,直接执行block(此时并没有切换线程进行调度)。

小结一下select 原理:

image.png

可以看出:

select 本身执行并不耗时,若最终没有数据返回则挂起等待,若是有数据返回则不会挂起协程。

我们从头再捋一下select 配合Channel 的原理:

image.png

虽然以Channel为例讲解了select 原理,实际上async等结合select 原理大致差不多,重点都是利用了协程的挂起/恢复做文章。

5. Select 注意事项

如果select有多个数据同时到达,select 默认会选择第一个数据,若想要随机选择数据,可做如下处理:

            var result = selectUnbiased<String> {
//监听是否有数据发送过来
receiveChannelZxing.onReceive { value ->
"zxing result $value"
}
}

想要知道select 还可以监听哪些数据,可查看该数据是否实现了SelectClauseX(X 表示0、1、2)。

以上即为Select 的原理及其使用,下篇将会进入协程的精华部分:Flow的运用,该部分内容较多,可能会分几篇分析,敬请期待。

本文基于Kotlin 1.5.3,文中完整Demo请点击


作者:小鱼人爱编程
链接:https://juejin.cn/post/7142083646822809607
来源:稀土掘金
著作权归作者所有。商业转载请联系作者获得授权,非商业转载请注明出处。

0 个评论

要回复文章请先登录注册