• Kotlin协程-select基础

    一、select是什么？

    select——>用于选择更快的结果。

    基于场景理解

    比如客户端要查询一个商品的详情。两个服务：缓存服务，速度快但信息可能是旧的；网络服务，速度慢但信息一定是最新的。

    如何实现上述逻辑：

    1.    runBlocking {
    2.         suspend fun getCacheInfo(productId: String): Product {
    3.             delay(100L)
    4.             return Product(productId, 8.9)
    5.         }
    6.  
    7.         suspend fun getNetworkInfo(productId: String): Product? {
    8.             delay(200L)
    9.             return Product(productId, 8.8)
    10.         }
    11.  
    12.         fun updateUI(product: Product) {
    13.             println("${product.productId} : ${product.price}")
    14.         }
    15.  
    16.         val startTime = System.currentTimeMillis()
    17.         val productId = "001"
    18.         val cacheInfo = getCacheInfo(productId)
    19.         if (cacheInfo != null) {
    20.             updateUI(cacheInfo)
    21.             println("Time cost: ${System.currentTimeMillis() - startTime}")
    22.         }
    23.  
    24.         val latestInfo = getNetworkInfo(productId)
    25.         if (latestInfo != null) {
    26.             updateUI(latestInfo)
    27.             println("Time cost: ${System.currentTimeMillis() - startTime}")
    28.         }
    29.     }
    30.  
    31.  
    32. 001 : 8.9
    33. Time cost: 113
    34. 001 : 8.8
    35. Time cost: 324

    上述程序分为四步：第一步：查询缓存信息；第二步：缓存服务返回信息，更新 UI；第三步：查询网络服务；第四步：网络服务返回信息，更新 UI。

    用户可以第一时间看到商品的信息，虽然它暂时会展示旧的信息，但由于我们同时查询了网络服务，旧缓存信息也马上会被替代成新的信息。但是可能存在一些问题：如果程序卡在了缓存服务，获取网络服务就会无法执行。是因为 getCacheInfo() 它是一个挂起函数，只有这个程序执行成功以后，才可以继续执行后面的任务。能否做到：两个挂起函数同时执行，谁返回的速度更快，就选择哪个结果。答案是使用select。

    1. runBlocking {
    2.  
    3.         suspend fun getCacheInfo(productId: String): Product {
    4.             delay(100L)
    5.             return Product(productId, 8.9)
    6.         }
    7.  
    8.         suspend fun getNetworkInfo(productId: String): Product {
    9.             delay(200L)
    10.             return Product(productId, 8.8)
    11.         }
    12.  
    13.         fun updateUI(product: Product) {
    14.             println("${product.productId} : ${product.price}")
    15.         }
    16.  
    17.         val startTime = System.currentTimeMillis()
    18.         val productId = "001"
    19.  
    20.         val product = select {
    21.  
    22.             async {
    23.                 getCacheInfo(productId)
    24.             }.onAwait {
    25.                 it
    26.             }
    27.  
    28.             async {
    29.                 getNetworkInfo(productId)
    30.             }.onAwait {
    31.                 it
    32.             }
    33.         }
    34.  
    35.         if (product != null) {
    36.             updateUI(product)
    37.             println("Time cost: ${System.currentTimeMillis() - startTime}")
    38.         }
    39.     }
    40.  
    41.  
    42. 001 : 8.9
    43. Time cost: 134
    44.  
    45. Process finished with exit code 0

    由于缓存的服务更快，所以，select 确实帮我们选择了更快的那个结果。我们的 select 可以在缓存服务出现问题的时候，灵活选择网络服务的结果。从而避免用户等待太长的时间，得到糟糕的体验。

    在上述代码中，用户大概率是会展示旧的缓存信息。但实际场景下，我们是需要进一步更新最新信息的。

    1. runBlocking {
    2.         suspend fun getCacheInfo(productId: String): Product {
    3.             delay(100L)
    4.             return Product(productId, 8.9)
    5.         }
    6.  
    7.         suspend fun getNetworkInfo(productId: String): Product {
    8.             delay(200L)
    9.             return Product(productId, 8.8)
    10.         }
    11.  
    12.         fun updateUI(product: Product) {
    13.             println("${product.productId} : ${product.price}")
    14.         }
    15.  
    16.         val startTime = System.currentTimeMillis()
    17.         val productId = "001"
    18.  
    19.         val cacheDeferred = async {
    20.             getCacheInfo(productId)
    21.         }
    22.         val latestDeferred = async {
    23.             getNetworkInfo(productId)
    24.         }
    25.  
    26.         val product = select {
    27.  
    28.             cacheDeferred.onAwait {
    29.                 it.copy(isCache = true)
    30.             }
    31.  
    32.             latestDeferred.onAwait {
    33.                 it.copy(isCache = false)
    34.             }
    35.         }
    36.  
    37.         if (product != null) {
    38.             updateUI(product)
    39.             println("Time cost: ${System.currentTimeMillis() - startTime}")
    40.         }
    41.  
    42.         if (product != null && product.isCache) {
    43.             val latest = latestDeferred.await() ?: return@runBlocking
    44.             updateUI(latest)
    45.             println("Time cost: ${System.currentTimeMillis() - startTime}")
    46.         }
    47.  
    48.     }
    49.  
    50.  
    51. 001 : 8.9
    52. Time cost: 124
    53. 001 : 8.8
    54. Time cost: 228
    55.  
    56. Process finished with exit code 0

    如果是多个服务的缓存场景呢？

    1.  runBlocking {
    2.  
    3.         val startTime = System.currentTimeMillis()
    4.         val productId = "001"
    5.  
    6.  
    7.         suspend fun getCacheInfo(productId: String): Product {
    8.             delay(100L)
    9.             return Product(productId, 8.9)
    10.         }
    11.  
    12.         suspend fun getCacheInfo2(productId: String): Product {
    13.             delay(50L)
    14.             return Product(productId, 8.85)
    15.         }
    16.  
    17.         suspend fun getNetworkInfo(productId: String): Product {
    18.             delay(200L)
    19.             return Product(productId, 8.8)
    20.         }
    21.  
    22.         fun updateUI(product: Product) {
    23.             println("${product.productId} : ${product.price}")
    24.         }
    25.  
    26.         val cacheDeferred = async {
    27.             getCacheInfo(productId)
    28.         }
    29.  
    30.         val cacheDeferred2 = async {
    31.             getCacheInfo2(productId)
    32.         }
    33.  
    34.         val latestDeferred = async {
    35.             getNetworkInfo(productId)
    36.         }
    37.  
    38.         val product = select {
    39.             cacheDeferred.onAwait {
    40.                 it.copy(isCache = true)
    41.             }
    42.  
    43.             cacheDeferred2.onAwait {
    44.                 it.copy(isCache = true)
    45.             }
    46.  
    47.             latestDeferred.onAwait {
    48.                 it.copy(isCache = true)
    49.             }
    50.         }
    51.         if (product != null) {
    52.             updateUI(product)
    53.             println("Time cost: ${System.currentTimeMillis() - startTime}")
    54.         }
    55.  
    56.         if (product != null && product.isCache) {
    57.             val latest = latestDeferred.await()
    58.             updateUI(latest)
    59.             println("Time cost: ${System.currentTimeMillis() - startTime}")
    60.         }
    61.  
    62.     }
    63.  
    64.  
    65. Log
    66.  
    67. 001 : 8.85
    68. Time cost: 79
    69. 001 : 8.8
    70. Time cost: 229
    71.  
    72. Process finished with exit code 0

    select 代码模式，可以提升程序的整体响应速度。

    二、select 和 Channel

    1. runBlocking {
    2.         val startTime = System.currentTimeMillis()
    3.         val channel1 = produce {
    4.             send(1)
    5.             delay(200L)
    6.             send(2)
    7.             delay(200L)
    8.             send(3)
    9.         }
    10.  
    11.         val channel2 = produce {
    12.             delay(100L)
    13.             send("a")
    14.             delay(200L)
    15.             send("b")
    16.             delay(200L)
    17.             send("c")
    18.         }
    19.  
    20.  
    21.  
    22.         channel1.consumeEach {
    23.             println(it)
    24.         }
    25.         channel2.consumeEach {
    26.             println(it)
    27.         }
    28.  
    29.         println("Time cost: ${System.currentTimeMillis() - startTime}")
    30.     }
    31.  
    32.  
    33. Log
    34.  
    35. 1
    36. 2
    37. 3
    38. a
    39. b
    40. c
    41. Time cost: 853
    42.  
    43. Process finished with exit code 0

    上述代码串行执行，可以使用select进行优化。

    1.   runBlocking {
    2.         val startTime = System.currentTimeMillis()
    3.         val channel1 = produce {
    4.             send(1)
    5.             delay(200L)
    6.             send(2)
    7.             delay(200L)
    8.             send(3)
    9.         }
    10.  
    11.         val channel2 = produce {
    12.             delay(100L)
    13.             send("a")
    14.             delay(200L)
    15.             send("b")
    16.             delay(200L)
    17.             send("c")
    18.         }
    19.  
    20.         suspend fun selectChannel(
    21.             channel1: ReceiveChannel<Int>,
    22.             channel2: ReceiveChannel<String>
    23.         ): Any {
    24.             return select {
    25.                 if (!channel1.isClosedForReceive) {
    26.                     channel1.onReceive {
    27.                         it.also {
    28.                             println(it)
    29.                         }
    30.                     }
    31.                 }
    32.  
    33.                 if (!channel2.isClosedForReceive) {
    34.                     channel2.onReceive {
    35.                         it.also {
    36.                             println(it)
    37.                         }
    38.                     }
    39.                 }
    40.             }
    41.         }
    42.  
    43.         repeat(6) {
    44.             selectChannel(channel1, channel2)
    45.         }
    46.         println("Time cost: ${System.currentTimeMillis() - startTime}")
    47.     }
    48.  
    49. Log
    50. 1
    51. a
    52. 2
    53. b
    54. 3
    55. c
    56. Time cost: 574
    57.  
    58. Process finished with exit code 0

    从代码执行结果可以发现程序的执行耗时有效减少。onReceive{} 是 Channel 在 select 当中的语法，当 Channel 当中有数据以后，它就会被回调，通过这个 Lambda，将结果传出去。 执行了 6 次 select，目的是要把两个管道中的所有数据都消耗掉。

    如果Channel1不生产数据了，程序会如何执行？

    1. runBlocking {
    2.         val startTime = System.currentTimeMillis()
    3.         val channel1 = produce {
    4.             delay(5000L)
    5.         }
    6.  
    7.         val channel2 = produce {
    8.             delay(100L)
    9.             send("a")
    10.             delay(200L)
    11.             send("b")
    12.             delay(200L)
    13.             send("c")
    14.         }
    15.  
    16.         suspend fun selectChannel(
    17.             channel1: ReceiveChannel<String>,
    18.             channel2: ReceiveChannel<String>
    19.         ): String = select {
    20.             channel1.onReceive {
    21.                 it.also {
    22.                     println(it)
    23.                 }
    24.             }
    25.             channel2.onReceive {
    26.                 it.also {
    27.                     println(it)
    28.                 }
    29.             }
    30.         }
    31.         repeat(3) {
    32.             selectChannel(channel1, channel2)
    33.         }
    34.         println("Time cost: ${System.currentTimeMillis() - startTime}")
    35.     }
    36.  
    37. Log
    38. a
    39. b
    40. c
    41. Time cost: 570
    42.  
    43. Process finished with exit code 0

     

    如果不知道Channel的个数，如何避免ClosedReceiveChannelException?

    使用：onReceiveCatching{} 

    1. runBlocking {
    2.         val startTime = System.currentTimeMillis()
    3.         val channel1 = produce {
    4.             delay(5000L)
    5.         }
    6.  
    7.         val channel2 = produce {
    8.             delay(100L)
    9.             send("a")
    10.             delay(200L)
    11.             send("b")
    12.             delay(200L)
    13.             send("c")
    14.         }
    15.  
    16.  
    17.         suspend fun selectChannel(
    18.             channel1: ReceiveChannel<String>,
    19.             channel2: ReceiveChannel<String>
    20.         ): String = select {
    21.             channel1.onReceiveCatching {
    22.                 it.getOrNull() ?: "channel1 is closed!"
    23.             }
    24.             channel2.onReceiveCatching {
    25.                 it.getOrNull() ?: "channel2 is closed!"
    26.             }
    27.         }
    28.  
    29.         repeat(6) {
    30.             val result = selectChannel(channel1, channel2)
    31.             println(result)
    32.         }
    33.  
    34.         println("Time cost: ${System.currentTimeMillis() - startTime}")
    35.     }
    36.  
    37. Log
    38. a
    39. b
    40. c
    41. channel2 is closed!
    42. channel2 is closed!
    43. channel2 is closed!
    44. Time cost: 584
    45.  
    46. Process finished with exit code 0

    得到所有结果以后，程序不会立即退出，因为 channel1 一直在 delay()。

    所以我们需要在6次repeat之后将channel关闭。

    1. runBlocking {
    2.         val startTime = System.currentTimeMillis()
    3.         val channel1 = produce {
    4.             delay(15000L)
    5.         }
    6.  
    7.         val channel2 = produce {
    8.             delay(100L)
    9.             send("a")
    10.             delay(200L)
    11.             send("b")
    12.             delay(200L)
    13.             send("c")
    14.         }
    15.  
    16.  
    17.         suspend fun selectChannel(
    18.             channel1: ReceiveChannel<String>,
    19.             channel2: ReceiveChannel<String>
    20.         ): String = select {
    21.             channel1.onReceiveCatching {
    22.                 it.getOrNull() ?: "channel1 is closed!"
    23.             }
    24.             channel2.onReceiveCatching {
    25.                 it.getOrNull() ?: "channel2 is closed!"
    26.             }
    27.         }
    28.  
    29.         repeat(6) {
    30.             val result = selectChannel(channel1, channel2)
    31.             println(result)
    32.         }
    33.  
    34.         channel1.cancel()
    35.         channel2.cancel()
    36.         println("Time cost: ${System.currentTimeMillis() - startTime}")
    37.     }
    38.  
    39. Log
    40. a
    41. b
    42. c
    43. channel2 is closed!
    44. channel2 is closed!
    45. channel2 is closed!
    46. Time cost: 612
    47.  
    48. Process finished with exit code 0

     Deferred、Channel 的 API:

    1.  
    2. public interface Deferred : CoroutineContext.Element {
    3.     public suspend fun join()
    4.     public suspend fun await(): T
    5.   
    6.     public val onJoin: SelectClause0
    7.     public val onAwait: SelectClause1
    8. }
    9.  
    10. public interface SendChannel<in E> 
    11.     public suspend fun send(element: E)
    12.  
    13.    
    14.     public val onSend: SelectClause2>
    15.  
    16. }
    17.  
    18. public interface ReceiveChannel<out E> {
    19.     public suspend fun receive(): E
    20.  
    21.     public suspend fun receiveCatching(): ChannelResult
    22.    
    23.     public val onReceive: SelectClause1
    24.     public val onReceiveCatching: SelectClause1>
    25. }

    当 select 与 Deferred 结合使用的时候，当并行的 Deferred 比较多的时候，你往往需要在得到一个最快的结果以后，去取消其他的 Deferred。

    通过 async 并发执行协程，也可以借助 select 得到最快的结果。

    1.  runBlocking {
    2.         suspend fun  fastest(vararg deferreds: Deferred<T>): T = select {
    3.             fun cancelAll() = deferreds.forEach {
    4.                 it.cancel()
    5.             }
    6.  
    7.             for (deferred in deferreds) {
    8.                 deferred.onAwait {
    9.                     cancelAll()
    10.                     it
    11.                 }
    12.             }
    13.         }
    14.  
    15.         val deferred1 = async {
    16.             delay(100L)
    17.             println("done1")
    18.             "result1"
    19.         }
    20.  
    21.         val deferred2 = async {
    22.             delay(200L)
    23.             println("done2")
    24.             "result2"
    25.         }
    26.  
    27.  
    28.         val deferred3 = async {
    29.             delay(300L)
    30.             println("done3")
    31.             "result3"
    32.         }
    33.  
    34.  
    35.         val deferred4 = async {
    36.             delay(400L)
    37.             println("done4")
    38.             "result4"
    39.         }
    40.  
    41.         val deferred5 = async {
    42.             delay(5000L)
    43.             println("done5")
    44.             "result5"
    45.         }
    46.  
    47.         val fastest = fastest(deferred1, deferred2, deferred3, deferred4, deferred5)
    48.         println(fastest)
    49.     }
    50.  
    51. Log
    52.  
    53. done1
    54. result1
    55.  
    56. Process finished with exit code 0

     

  • 相关阅读:
    09.06app端自动化
    热力学相关的两个定律
    鸿蒙开发实例：【配置OpenHarmony SDK】
    Git 学习笔记
    【优化方案】Java 将字符串中的星号替换为0-9中的数字，并返回所有可能的替换结果
    2022.7.29好题选讲(计数专题)
    猿创征文｜DEM分析分层重分类
    virtio代码分析(一)-qemu部分
    老子云平台会员专业又有性价比！
    Java 类和对象
  • 原文地址：https://blog.csdn.net/zhangying1994/article/details/127485681