Kotlin select使用方法介绍
作者：且听真言 发布时间：2022-05-28 19:34:27
标签：Kotlin,select
一、select是什么
select——>用于选择更快的结果。
基于场景理解
比如客户端要查询一个商品的详情。两个服务：缓存服务，速度快但信息可能是旧的；网络服务，速度慢但信息一定是最新的。
如何实现上述逻辑：
runBlocking {
suspend fun getCacheInfo(productId: String): Product {
delay(100L)
return Product(productId, 8.9)
}
suspend fun getNetworkInfo(productId: String): Product? {
delay(200L)
return Product(productId, 8.8)
}
fun updateUI(product: Product) {
println("${product.productId} : ${product.price}")
}
val startTime = System.currentTimeMillis()
val productId = "001"
val cacheInfo = getCacheInfo(productId)
if (cacheInfo != null) {
updateUI(cacheInfo)
println("Time cost: ${System.currentTimeMillis() - startTime}")
}
val latestInfo = getNetworkInfo(productId)
if (latestInfo != null) {
updateUI(latestInfo)
println("Time cost: ${System.currentTimeMillis() - startTime}")
}
}
001 : 8.9
Time cost: 113
001 : 8.8
Time cost: 324
上述程序分为四步：第一步：查询缓存信息；第二步：缓存服务返回信息，更新 UI；第三步：查询网络服务；第四步：网络服务返回信息，更新 UI。
用户可以第一时间看到商品的信息，虽然它暂时会展示旧的信息，但由于我们同时查询了网络服务，旧缓存信息也马上会被替代成新的信息。但是可能存在一些问题：如果程序卡在了缓存服务，获取网络服务就会无法执行。是因为 getCacheInfo() 它是一个挂起函数，只有这个程序执行成功以后，才可以继续执行后面的任务。能否做到：两个挂起函数同时执行，谁返回的速度更快，就选择哪个结果。答案是使用select。
runBlocking {
suspend fun getCacheInfo(productId: String): Product {
delay(100L)
return Product(productId, 8.9)
}
suspend fun getNetworkInfo(productId: String): Product {
delay(200L)
return Product(productId, 8.8)
}
fun updateUI(product: Product) {
println("${product.productId} : ${product.price}")
}
val startTime = System.currentTimeMillis()
val productId = "001"
val product = select<Product?> {
async {
getCacheInfo(productId)
}.onAwait {
it
}
async {
getNetworkInfo(productId)
}.onAwait {
it
}
}
if (product != null) {
updateUI(product)
println("Time cost: ${System.currentTimeMillis() - startTime}")
}
}
001 : 8.9
Time cost: 134
Process finished with exit code 0
由于缓存的服务更快，所以，select 确实帮我们选择了更快的那个结果。我们的 select 可以在缓存服务出现问题的时候，灵活选择网络服务的结果。从而避免用户等待太长的时间，得到糟糕的体验。
在上述代码中，用户大概率是会展示旧的缓存信息。但实际场景下，我们是需要进一步更新最新信息的。
runBlocking {
suspend fun getCacheInfo(productId: String): Product {
delay(100L)
return Product(productId, 8.9)
}
suspend fun getNetworkInfo(productId: String): Product {
delay(200L)
return Product(productId, 8.8)
}
fun updateUI(product: Product) {
println("${product.productId} : ${product.price}")
}
val startTime = System.currentTimeMillis()
val productId = "001"
val cacheDeferred = async {
getCacheInfo(productId)
}
val latestDeferred = async {
getNetworkInfo(productId)
}
val product = select<Product?> {
cacheDeferred.onAwait {
it.copy(isCache = true)
}
latestDeferred.onAwait {
it.copy(isCache = false)
}
}
if (product != null) {
updateUI(product)
println("Time cost: ${System.currentTimeMillis() - startTime}")
}
if (product != null && product.isCache) {
val latest = latestDeferred.await() ?: return@runBlocking
updateUI(latest)
println("Time cost: ${System.currentTimeMillis() - startTime}")
}
}
001 : 8.9
Time cost: 124
001 : 8.8
Time cost: 228
Process finished with exit code 0
如果是多个服务的缓存场景呢？
runBlocking {
val startTime = System.currentTimeMillis()
val productId = "001"
suspend fun getCacheInfo(productId: String): Product {
delay(100L)
return Product(productId, 8.9)
}
suspend fun getCacheInfo2(productId: String): Product {
delay(50L)
return Product(productId, 8.85)
}
suspend fun getNetworkInfo(productId: String): Product {
delay(200L)
return Product(productId, 8.8)
}
fun updateUI(product: Product) {
println("${product.productId} : ${product.price}")
}
val cacheDeferred = async {
getCacheInfo(productId)
}
val cacheDeferred2 = async {
getCacheInfo2(productId)
}
val latestDeferred = async {
getNetworkInfo(productId)
}
val product = select<Product?> {
cacheDeferred.onAwait {
it.copy(isCache = true)
}
cacheDeferred2.onAwait {
it.copy(isCache = true)
}
latestDeferred.onAwait {
it.copy(isCache = true)
}
}
if (product != null) {
updateUI(product)
println("Time cost: ${System.currentTimeMillis() - startTime}")
}
if (product != null && product.isCache) {
val latest = latestDeferred.await()
updateUI(latest)
println("Time cost: ${System.currentTimeMillis() - startTime}")
}
}
Log
001 : 8.85
Time cost: 79
001 : 8.8
Time cost: 229
Process finished with exit code 0
select 代码模式，可以提升程序的整体响应速度。
二、select和Channel
runBlocking {
val startTime = System.currentTimeMillis()
val channel1 = produce {
send(1)
delay(200L)
send(2)
delay(200L)
send(3)
}
val channel2 = produce {
delay(100L)
send("a")
delay(200L)
send("b")
delay(200L)
send("c")
}
channel1.consumeEach {
println(it)
}
channel2.consumeEach {
println(it)
}
println("Time cost: ${System.currentTimeMillis() - startTime}")
}
Log
1
2
3
a
b
c
Time cost: 853
Process finished with exit code 0
上述代码串行执行，可以使用select进行优化。
runBlocking {
val startTime = System.currentTimeMillis()
val channel1 = produce {
send(1)
delay(200L)
send(2)
delay(200L)
send(3)
}
val channel2 = produce {
delay(100L)
send("a")
delay(200L)
send("b")
delay(200L)
send("c")
}
suspend fun selectChannel(
channel1: ReceiveChannel<Int>,
channel2: ReceiveChannel<String>
): Any {
return select<Any> {
if (!channel1.isClosedForReceive) {
channel1.onReceive {
it.also {
println(it)
}
}
}
if (!channel2.isClosedForReceive) {
channel2.onReceive {
it.also {
println(it)
}
}
}
}
}
repeat(6) {
selectChannel(channel1, channel2)
}
println("Time cost: ${System.currentTimeMillis() - startTime}")
}
Log
1
a
2
b
3
c
Time cost: 574
Process finished with exit code 0
从代码执行结果可以发现程序的执行耗时有效减少。onReceive{} 是 Channel 在 select 当中的语法，当 Channel 当中有数据以后，它就会被回调，通过这个 Lambda，将结果传出去。执行了 6 次 select，目的是要把两个管道中的所有数据都消耗掉。
如果Channel1不生产数据了，程序会如何执行？
runBlocking {
val startTime = System.currentTimeMillis()
val channel1 = produce<String> {
delay(5000L)
}
val channel2 = produce<String> {
delay(100L)
send("a")
delay(200L)
send("b")
delay(200L)
send("c")
}
suspend fun selectChannel(
channel1: ReceiveChannel<String>,
channel2: ReceiveChannel<String>
): String = select<String> {
channel1.onReceive {
it.also {
println(it)
}
}
channel2.onReceive {
it.also {
println(it)
}
}
}
repeat(3) {
selectChannel(channel1, channel2)
}
println("Time cost: ${System.currentTimeMillis() - startTime}")
}
Log
a
b
c
Time cost: 570
Process finished with exit code 0
如果不知道Channel的个数，如何避免ClosedReceiveChannelException?
使用：onReceiveCatching{}
runBlocking {
val startTime = System.currentTimeMillis()
val channel1 = produce<String> {
delay(5000L)
}
val channel2 = produce<String> {
delay(100L)
send("a")
delay(200L)
send("b")
delay(200L)
send("c")
}
suspend fun selectChannel(
channel1: ReceiveChannel<String>,
channel2: ReceiveChannel<String>
): String = select<String> {
channel1.onReceiveCatching {
it.getOrNull() ?: "channel1 is closed!"
}
channel2.onReceiveCatching {
it.getOrNull() ?: "channel2 is closed!"
}
}
repeat(6) {
val result = selectChannel(channel1, channel2)
println(result)
}
println("Time cost: ${System.currentTimeMillis() - startTime}")
}
Log
a
b
c
channel2 is closed!
channel2 is closed!
channel2 is closed!
Time cost: 584
Process finished with exit code 0
得到所有结果以后，程序不会立即退出，因为 channel1 一直在 delay()。
所以我们需要在6次repeat之后将channel关闭。
runBlocking {
val startTime = System.currentTimeMillis()
val channel1 = produce<String> {
delay(15000L)
}
val channel2 = produce<String> {
delay(100L)
send("a")
delay(200L)
send("b")
delay(200L)
send("c")
}
suspend fun selectChannel(
channel1: ReceiveChannel<String>,
channel2: ReceiveChannel<String>
): String = select<String> {
channel1.onReceiveCatching {
it.getOrNull() ?: "channel1 is closed!"
}
channel2.onReceiveCatching {
it.getOrNull() ?: "channel2 is closed!"
}
}
repeat(6) {
val result = selectChannel(channel1, channel2)
println(result)
}
channel1.cancel()
channel2.cancel()
println("Time cost: ${System.currentTimeMillis() - startTime}")
}
Log
a
b
c
channel2 is closed!
channel2 is closed!
channel2 is closed!
Time cost: 612
Process finished with exit code 0
Deferred、Channel 的 API:
public interface Deferred : CoroutineContext.Element {
public suspend fun join()
public suspend fun await(): T
public val onJoin: SelectClause0
public val onAwait: SelectClause1<T>
}
public interface SendChannel<in E>
public suspend fun send(element: E)
public val onSend: SelectClause2<E, SendChannel<E>>
}
public interface ReceiveChannel<out E> {
public suspend fun receive(): E
public suspend fun receiveCatching(): ChannelResult<E>
public val onReceive: SelectClause1<E>
public val onReceiveCatching: SelectClause1<ChannelResult<E>>
}
当 select 与 Deferred 结合使用的时候，当并行的 Deferred 比较多的时候，你往往需要在得到一个最快的结果以后，去取消其他的 Deferred。
通过 async 并发执行协程，也可以借助 select 得到最快的结果。
runBlocking {
suspend fun <T> fastest(vararg deferreds: Deferred<T>): T = select {
fun cancelAll() = deferreds.forEach {
it.cancel()
}
for (deferred in deferreds) {
deferred.onAwait {
cancelAll()
it
}
}
}
val deferred1 = async {
delay(100L)
println("done1")
"result1"
}
val deferred2 = async {
delay(200L)
println("done2")
"result2"
}
val deferred3 = async {
delay(300L)
println("done3")
"result3"
}
val deferred4 = async {
delay(400L)
println("done4")
"result4"
}
val deferred5 = async {
delay(5000L)
println("done5")
"result5"
}
val fastest = fastest(deferred1, deferred2, deferred3, deferred4, deferred5)
println(fastest)
}
Log
done1
result1
Process finished with exit code 0
来源：https://blog.csdn.net/zhangying1994/article/details/127485681
0
投稿猜你喜欢
01.点明观点 C#中，非托管资源使用之后必须释放，而using()是使用非托管资源的最佳方式，可以确保资源在代码块结束之后被正确- 1,compareTo(Object o)方法是java.lang.Comparable<T>接口中的方法，当需要对某个类的对象
- 参考内容：深入理解Java虚拟机(JVM高级特性与最佳实践) ——周志明老师尚硅谷深入理解JVM教学视频——宋红康老师在本文展开前，读者需要
- 前言通过ioctl跟binder驱动交互，实现以最快的方式唤醒新的保活服务，最大程度防止保活失败。同时，我也将跟您分享，我是怎么做到在不甚了
- 本文实例讲述了C#实现中英文混合字符串截取的方法，是C#字符串操作中非常常用的一个方法。分享给大家供大家参考之用。具体方法如下：具体功能代码
- 一、什么是代理?指为一个目标对象提供一个代理对象, 并由代理对象控制对目标对象的引用. 使用代理对象, 是为了在不修改目标对象的基础上,增强
- 首先，通过代码定义一个委托和下面三个示例将要调用的方法：public delegate int AddHandler(int a,int b
- 本文较为详细的分析了C#读取二进制文件方法。分享给大家供大家参考。具体分析如下：当想到所有文件都转换为 XML时，确实是一件好事。但是，这并
- 一、事务的基本原理Spring事务的本质其实就是数据库对事务的支持，没有数据库的事务支持，spring是无法提供事务功能的。对于纯JDBC操
- 大致分为以下几个方面：一些查询指令整理使用SQL语句进行特殊查询检测表字段是否存在数据库升级数据库表字段赋初始值一、查询指令整理1.链式执行
- 使用DOM4J方式生成XML文件的步骤如下：引入JAR包通过DocumentHelper类的createDocument()创建Docume
- 创建一个maven项目导入springboot依赖，注意底 * 释部分<?xml version="1.0&quo
- 文章来源：aspcn 作者：孙雯重复和并发服务器这个应用程序被当作一个重复的服务器.因为它只有在处理完一个进程以后才会接受另一个连接.更多的
- 简单实现了下：import javax.crypto.BadPaddingException;import javax.crypto.Cip
- 简单看一下描述，例子最重要。1、getPath()：返回定义时的路径，（就是你写什么路径，他就返回什么路径）2、getAbsolutePat
- 一、前言写今天这篇文章的缘由，其实是来自于前段时间和粉丝的一个聊天，最近他打算参加游戏创作大赛，问我需要准备学习什么知识，以及参加比赛的注意
- Android 6本文根据我个人的开发经验，总结了从 Android 6 - Android 13 重要的行为变更。当然，这不是 Andro
- 目录业务场景：解决方案：总结业务场景：一个用户数据接口，要求在20ms内返回数据，它的调用逻辑复杂，关联接口多，需要从3个接口汇总数据，这些
- 概述使用this()或target()可绑定被代理对象实例，在通过类实例名绑定对象时，还依然具有原来连接点匹配的功能，只不过类名是通过增强方
- 测试Spring Boot定时任务冲突时，使用的线程数量引入依赖：Spring Boot 2.6.1 <dependency>
