JOOQ 3.17.x：如何结合R2DBC调用批量操作方法？

问题背景 #

在jOOQ搭配R2DBC使用时，SELECT查询不能调用.execute()这类阻塞方法，得用.asFlow().toList()这类响应式方式获取结果。但尝试用batchUpdate执行批量操作时，调用.execute()会触发错误：

batchUpdate(myRecords).execute()

错误信息：

Attempt to execute a blocking method (e.g. Query.execute() or ResultQuery.fetch()) when only an R2BDC ConnectionFactory was configured. jOOQ's RowCountQuery and ResultQuery extend Publisher, which allows for reactive streams implementations to subscribe to the results of a jOOQ query. Simply embed your query in the stream, e.g. using Flux.from(query).

解决方案 #

当然可以结合R2DBC执行批量操作，关键是放弃阻塞的.execute()，改用响应式API触发执行：

• batchUpdate返回的Batch对象本身实现了响应式流的Publisher接口，直接用响应式框架的方法订阅即可。
• 在Kotlin环境中，推荐转成Kotlin Flow处理：
  1. 仅执行批量更新，不需要返回结果：

  batchUpdate(myRecords).asFlow().collect()
  

  2. 需要获取每条更新的影响行数：

  val updateRowCounts = batchUpdate(myRecords).asFlow().toList()
  

• 如果使用Spring WebFlux等基于Reactor的框架，也可以用Flux.from()包装后执行：

  Flux.from(batchUpdate(myRecords)).subscribe()
  

核心逻辑是：R2DBC是响应式驱动，jOOQ针对这种场景把所有操作对象都适配成了响应式流类型，只要遵循响应式的执行方式，就能避免阻塞方法的报错。

内容的提问来源于stack exchange，提问作者Soichi Takamura