jOOQ 是否与 Kotlin 协程配合得很好？答案

【问题标题】：Does jOOQ play nicely with Kotlin coroutines?jOOQ 是否与 Kotlin 协程配合得很好？
【发布时间】：2021-10-25 04:23:49
【问题描述】：

Kotlin coroutines 和“挂起函数”使程序员可以轻松地等待 I/O 的结果而无需停止线程（线程在 I/O 之前还有其他工作要做已完成）。

jOOQ 是 Java 优先产品，用于以类型安全的方式编写和执行 SQL，但它本身并没有明确使用 Kotlin 协程。

是否可以从 Kotlin 协程范围调用 jOOQ 以获得易于编写和线程高效的优势？

suspend fun myQuery() {
  return dsl.select()
  // .etc()
  .fetch()  // <- could this be a 'suspend' call?
}

【问题讨论】：

标签： kotlin-coroutines jooq nonblocking

【解决方案1】：

答：是的。

因为 org.jooq.ResultQuery<R extends Record> 有一个 @NotNull CompletionStage<Result<R>> fetchAsync()` 方法，它连接到 Java (JDK 8+) 的“futures”机器中。

与 kotlinx-coroutines-jdk8 提供了一个 bunch of extension methods 用于在 Kotlin 挂起函数和 JDK 8+ futures 之间进行适配。

因此我们可以做到：

import kotlinx.coroutines.future.await
...
suspend fun myQuery() {
  return dsl.select()
  //.etc()
  .fetchAsync()
  .await()    // <- IS  a suspending call !!!
}

需要注意的是，ResultQuery 上有一堆重载的 fetchX() 方法，它们为同步调用提供了很多实用程序，但对于 fetchAsync()，它们没有类似的重载。这就是 Kotlin 程序员可能希望熟悉 Java 期货机制的地方：使用 CompletionStage<T> 上的 thenApply {} method 可以异步完成任何类型的操作。例如，映射结果：

suspend fun myQuery() {
  return dsl.select()
  //.etc()
  .fetchAsync()
  .thenApply { it.map(mapping(::Film)) }  // <- done as part of the 'suspend'
  .await()
}

虽然在暂停后这样做应该没问题：

suspend fun myQuery() {
  val records = dsl.select()
  //.etc()
  .fetchAsync()
  .await()

  return records.map(mapping(::Film))
}

【讨论】：

如您上一个问题所述，CompletionStage 有望成为non-blocking in the near future, if supplied with an R2DBC connection
@LukasEder 是的，尽管 r2dbc 还没有在幕后用于实际的非阻塞。任何使用 JDBC 的人都有阻塞行为，所以使用 jOOQ 并不会丢失。只有那些愿意将他们的代码形状扭曲为响应式风格，并且愿意忍受 r2dbc 的不成熟的人，才会“过得更好”。但是有了 jOOQ，我可以耐心地等待这些收获，同时按照它应该的方式编写我的应用程序。
@DavidBullock 在 JDBC 阻塞模式下使用此 API 时，您对为 jooq 配置 executorProvider 线程池的最佳方式有任何指导吗？作为参考，我使用 ktor 作为我的服务器（基于协程），因此我没有为阻塞数据库操作配置现有线程池。

【解决方案2】：

JOOQ 添加了 R2dbc 支持，ResultQuery 是一个 ReactiveStreams Publisher。

~~但我没有找到对 Kotlin Coroutines 的直接支持，~~ Jooq 3.17 在jooq-kotlin 模块中提供了官方 Koltin Coroutines 支持。

目前最简单的方法是使用 Reactor Kotlin 扩展将 Reactor API 转换为 Kotlin Coroutines API。

  Flux.from(
     ctx.select()
     ...//etc
     ...// do not call fetch
  ) 
  .asFlow()

更新：切换到3.17或更高版本时，直接使用fetchAwait。已添加一系列xxxAwait。

ctx.select()
     ...//etc
     .fetchAwait()

【讨论】：