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

Question

我正在使用没有响应式 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
    }

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

我该如何解决？

Answer 1

TL;DR 将async 与用于卸载阻塞IO 的自定义调度程序（例如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，在 coroutineScope 中使用 async {...} 似乎没有什么不同，并且执行时间与同步代码相当。

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

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

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

您为什么要使用这种方法？

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

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

这是个好主意吗？

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

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

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

Answer 2

该代码是并行工作的：

    @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
        }
    }

Answer 3

也许根本原因是：

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

解决方案：

• 使用使用 java.nio 的 http-client

Answer 4

• 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()}!")
        }
    }