来自阻塞代码的异步/等待 kotlin 协程

Question

我使用的是 Spring boot，没有响应式 Web。

我尝试使用 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
    }

但这段代码是按顺序运行的。

我该如何修复它？

Answer 1

TL;DR与自定义调度程序一起 使用async，例如Dispatchers.IO旨在卸载阻塞 IO 的调度程序。

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

更新：我在Kotlin 协程 slack 频道 中被告知，我可以使用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 {...}在 the 内使用coroutineScope似乎没有什么区别，并且执行时间与同步代码相同。

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

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

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

为什么您可能想要使用此方法？

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

如果您要开始一个新项目，请不要使用RestTemplate. 相反，最好将WebFlux 与 Kotlin 中的协程一起使用，如本文所述。

这是一个好主意吗？

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

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

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