如何在 webflux 上正确使用带有协程的挂起函数？

Question

我是响应式编程的新手，因为我过去已经在 spring-web 中使用过 kotlin，所以我决定在我正在从事的这个新项目中使用 spring-webflux。然后我发现了 Mono 和 Flux api 并决定使用 spring-data-r2dbc 来保持完整的反应堆栈（我知道我不知道这个新项目离满足所有反应还有多远期望，我这样做是为了学习一个新工具，而不是因为这是这个新工具的完美场景）

然后我注意到我可以用 kotlin 的原生协程替换来自 webflux 的所有反应流 api。我还选择了协程来简单地学习并减少 'external frameworky' 代码

我的应用程序非常简单（它是一个 url 缩短器）：

1. parse some url out of http request's body into 3 parts
2. exchange each part to its postgres id on each respective table
3. concat these 3 ids into a new url, sending an 200 http response with this new url

我的反应控制器是

@Configuration
class UrlRouter {

    @Bean
    fun urlRoutes(
        urlHandler: UrlHandler,
        redirectHandler: RedirectHandler
    ) = coRouter {
        POST("/e", urlHandler::encode)
        GET("/{*url}", redirectHandler::redirect)
    }
}

如你所想，UrlHandler 负责上面编号的步骤，RedirectHandler 做相反的事情：接收编码的 url，它重定向到右边 url 在 1 号收到。

问题 1： 检查 coRouter，我假设对于每个 http 调用，spring 将启动一个新协程来解决该调用（反对到传统 spring-web 上的新线程），并且其中的每一个都可以创建并依赖于其他几个子协程。这是正确的吗？这个层次结构存在吗？

这是我的 UrlHandler 片段：

@Component
class UrlHandler(
    private val cache: CacheService,
    @Value("${redirect-url-prefix}") private val prefix: String
) {

    companion object {
        val mapper = jacksonObjectMapper()
    }

    suspend fun encode(serverRequest: ServerRequest): ServerResponse =
        try {
            val bodyMap: Map<String, String> = mapper.readValue(serverRequest.awaitBody<String>())
            // parseUrl being a string extension function just splitting 
            // that could throw IndexOutOfBoundsException
            val (host, path, query) = bodyMap["url"]!!.parseUrl()

            val hostId: Long = cache.findIdFromHost(host)
            val pathId: Long? = cache.findIdFromPath(path)
            val queryId: Long? = cache.findIdFromQuery(query)

            val encodedUrl = "$prefix/${someOmmitedStringConcatenation(hostId, pathId, queryId)}"
            ok().bodyValueAndAwait(mapOf("url" to encodedUrl))
        } catch (e: IndexOutOfBoundsException) {
            ServerResponse.badRequest().buildAndAwait()
        }

所有三个 findIdFrom*** 调用都尝试检索现有 ID，如果不存在，则从 postgres 序列中保存新实体和 return 新 ID。这是通过 CoroutineCrudRepository 接口完成的。由于我的，所有 3 findIdFrom*** 也 suspend:

@Repository
interface HostUrlRepo : CoroutineCrudRepository<HostUrl, Long> {
    suspend fun findByHost(host: String): HostUrl?
}

问题 2： 寻找 here 我找到了 either invoke reactive query methods or have native suspended functions。自从我阅读了 methods should always suspend，我决定继续使用 suspend。这是bad/wrong吗？

这 3 个 findIdFrom*** 是独立的，可以并行调用运行，然后仅在 someOmmitedStringConcatenation 我应该等待任何未完成的调用来实际构建我的编码 url

问题 3： 因为每个方法都有 suspend 修饰符，所以它将运行与传统的命令式顺序范例完全一样（浪费任何好处并行编程) ?

问题 4：这是协程使用的有效场景吗？如果是这样，我应该如何更改我的代码以最适合上面我想要的并行性？

我为问题 4 找到的可能解决方案：

问题 4.1： source 1 在每个 findIdFrom*** 中用 withContext(Dispatchers.IO){ /*actual code here*/ } 包装它然后在 encode 函数上：

coroutineScope {
    val hostIdDeferred = async { findIdFrom***() }
    val pathIdDeferred = async { findIdFrom***() }
    val queryIdDeferred = async { findIdFrom***() }
}

当我想使用它们时，只需使用hostIdDeferred.await() 来获取值。如果我在新的子协程中使用 Dispatchers.IO 范围到运行代码，为什么 coroutineScope 是必要的？这是正确的方法吗，为新的协程子指定一个范围，然后使用 coroutineScope 来获得一个延迟的 val？

问题 4.2: source 2 val resultOne = Async(Dispatchers.IO) { function1() } Intellij 无法 recognize/import 任何 Async 表达式。我如何使用这个以及它与之前的有何不同？

我愿意改进和澄清关于这个问题的任何观点

Answer 1

我会尽量回答你的一些问题：

q2：没有，没什么问题。暂停方法可以一直传播回控制器。如果您的控制器是反应式的，即如果您将 RSocket 与 org.springframework.messaging.handler.annotation.MessageMapping 一起使用，那么甚至控制器方法也可以被挂起。
q3：对，但是每个方法还是你的源码简单多了
q4.2：我不认为该网站是值得信赖的来源。有一个带有示例的官方文档：async

如何在 webflux 上正确使用带有协程的挂起函数？

How to correctly use suspend functions with coroutines on webflux?

reactive-programming

spring-webflux

kotlin-coroutines