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

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

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

然后我注意到我可以用 kotlin 的原生协程替换来自 webflux 的所有反应流 API。我也选择协程只是为了学习和减少“外部框架”代码

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

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，它重定向到在编号 1 上接收到的正确 url。

问题 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 序列返回新 ID。这是由 CoroutineCrudRepository 接口完成的。由于我的 methods should always suspend，所有 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。这有什么不好/错误吗？

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

问题 3：因为每个方法都有 suspend 修饰符，所以它会像传统的命令式顺序范式一样运行（浪费并行编程的任何好处）？

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

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

问题 4.1： 每个 findIdFrom*** 内的 source 1 用 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 无法识别/导入任何 Async 表达式。我该如何使用这个，它与之前的有什么不同？

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

解决方法

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

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

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

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

解决方法

相关推荐