异步/等待Kotlin协程阻止代码

如何解决异步/等待Kotlin协程阻止代码

我正在使用没有响应式Web的Spring Boot。

我尝试使用Kotlin协程运行一些异步请求

    @GetMapping
    fun test(): Message {
        val restTemplate = RestTemplate()
        return runBlocking {
            val hello = async { hello(restTemplate) }
            val world = async { world(restTemplate) }
            Message("${hello.await()} ${world.await()}!")
        }
    }

    private suspend fun world(restTemplate: RestTemplate): String {
        logger.info("Getting WORLD")
        return restTemplate.getForEntity("http://localhost:9090/world",World::class.java).body!!.payload
    }

    private suspend fun hello(restTemplate: RestTemplate): String {
        logger.info("Getting HELLO")
        return restTemplate.getForEntity("http://localhost:9090/hello",Hello::class.java).body!!.payload
    }

但是此代码按顺序运行。

我该如何解决？

解决方法

TL;DR 将 async 与用于卸载阻塞 IO 的自定义 Dispatcher（例如 Dispatchers.IO）一起使用。

val hello = async(Dispatchers.IO) { hello(restTemplate) }
val world = async(Dispatchers.IO) { world(restTemplate) }

更新： 我在 Kotlin coroutines slack channel 中被告知，我可以使用 async(Dispatchers.IO) 而不是使用 async + withContext(Dispatchers.IO)。

~~我采用了@Sergey Nikulitsa 代码并创建了一个扩展函数，该函数采用带有接收器的 lambda（类似于 async）来组合 async 和 withContext(Dispatches.IO)。~~

import kotlinx.coroutines.*

fun <T> CoroutineScope.myRestTemplateAsync(
    start: CoroutineStart = CoroutineStart.DEFAULT,block: suspend CoroutineScope.() -> T
): Deferred<T> {

    return async(Dispatchers.IO,start) {
        block() 
    }
}

然后它可以像这样在你的代码中使用：


@GetMapping
fun test(): Message {
    val restTemplate = RestTemplate()
    return runBlocking {
        val hello = myRestTemplateAsync { hello(restTemplate) }
        val world = myRestTemplateAsync { world(restTemplate) }
        Message("${hello.await()} ${world.await()}!")
    }
}

private suspend fun world(restTemplate: RestTemplate): String {
    logger.info("Getting WORLD")
    return restTemplate.getForEntity("http://localhost:9090/world",World::class.java).body!!.payload
}

private suspend fun hello(restTemplate: RestTemplate): String {
    logger.info("Getting HELLO")
    return restTemplate.getForEntity("http://localhost:9090/hello",Hello::class.java).body!!.payload
}

初步结果

此时，我只是在试验这个方法，我只使用 Spring WebMVC 和 RestTemplate 进行 5+ 次调用。

myRestTemplateAsync 扩展函数与同步对应函数相比，执行时间持续减少了 30% 到 50%。

为什么这比使用 `async { }` 更有效？

特别是对于 RestTemplate，在 async {...} 中使用 coroutineScope 似乎没有什么区别，并且执行时间与同步代码相当。

此外，查看分析器中的线程，在单独使用 async 时没有创建“Dispatcher Workers”。这让我相信 RestTemplate 的每请求线程模型阻塞了整个线程。

当在 async 中指定新的调度程序时，它将协程（和函数 block）的执行转移到 Dispatchers.IO 线程池中的一个新线程。

在这种情况下，代码块应包含 RestTemplate 调用（单个调用）。据我所知，这可以防止 RestTemplate 阻塞原始上下文。

为什么要使用这种方法？

如果您一直在大型项目中使用 RestTemplate（每请求线程模型），那么仅将其替换为非阻塞客户端（如 WebClient）可能是一项艰巨的任务。这样，您就可以继续使用大部分代码，只需在代码中可以异步进行多次调用的区域添加 myRestTemplateAsync。

如果您要开始一个新项目，请不要使用 RestTemplate。最好使用 WebFlux with coroutines in Kotlin as explained in this article。

这是个好主意吗？

目前，我没有足够的信息来说明这一点。我希望进行更广泛的测试和评估：

负载下的内存消耗
负载下可能耗尽线程池
异常是如何传播和处理的

如果您对为什么这可能是一个好主意或可能不是一个好主意有任何意见，请在下面发表。

runBlocking：旨在将常规的阻塞代码桥接到以挂起方式编写的库中，以用于主要功能和测试中。
在这里，我们使用coroutineScope方法创建一个CoroutineScope。此功能设计用于并行分解工作。当此范围内的任何子协程失败时，该范围将失败，并且所有其他子级都将被取消。
因为coroutineScope是暂停函数，所以我们将fun test()标记为suspend fun（只允许从协程或另一个暂停函数调用暂停函数）。通过使用CoroutineScope对象，我们可以调用async和launch来启动协程

  @GetMapping
  suspend fun test(): Message {
        val restTemplate = RestTemplate()
        return coroutineScope {
            val hello = async { hello(restTemplate) }
            val world = async { world(restTemplate) }
            Message("${hello.await()} ${world.await()}!")
        }
    }

也许根本原因是：

restTemplate使用java.io（不是java.nio）
restTemplate阻止当前线程，直到获得HTTP响应
协程魔术在这种情况下不起作用

解决方案：

使用使用java.nio的http客户端

该代码可以并行工作：

    @GetMapping
    fun test(): Message {
        val restTemplate = RestTemplate()
        return runBlocking {
            val hello = async { hello(restTemplate) }
            val world = async { world(restTemplate) }
            Message("${hello.await()} ${world.await()}!")
        }
    }

    private suspend fun world(restTemplate: RestTemplate): String {
        logger.info("Getting WORLD")
        return withContext(Dispatchers.IO) {
            restTemplate.getForEntity("http://localhost:9090/world",World::class.java).body!!.payload
        }
    }

    private suspend fun hello(restTemplate: RestTemplate): String {
        logger.info("Getting HELLO")
        return withContext(Dispatchers.IO) {
            restTemplate.getForEntity("http://localhost:9090/hello",Hello::class.java).body!!.payload
        }
    }