在 Kotlin 序列的嵌套对象中使用 Yield

Question

我想RowCallbackHandler通过 Kotlin 流式传输由 Spring JDBC 捕获的结果对象Sequence。

代码基本上是这样的：

fun findManyObjects(): Sequence<Thing> = sequence {
    val rowHandler = object : RowCallbackHandler {
        override fun processRow(resultSet: ResultSet) {
            val thing = // create from resultSet
            yield(thing) // ERROR! No coroutine scope
        }
    }
    jdbcTemplate.query("select * from ...", rowHandler)
}
    

但我收到编译错误：

挂起函数只能在协程体内调用。

然而，这个“协程体”确实应该存在，因为整个块都被包装在一个sequence构建器中。但它似乎不适用于嵌套对象。

显示它不使用嵌套对象进行编译的最小示例：

// compiles
sequence {
    yield(1)
}

// doesn't compile
sequence {
   object {
        fun doit() {
            yield(1) // Suspension functions can be called only within coroutine body.
        }
    }
}

如何将对象从 传递ResultSet到Sequence？

Answer 1

用于Flow异步数据流

无法在对象yield内部调用的原因RowCallbackHandler有两个。

1. 该processRow函数不是挂起函数（也不可能是，因为它是在 Java 中声明并由 Java 调用的）。类似的挂起函数yield只能由另一个挂起函数调用。
2. 当构建器返回时，序列总是结束sequence { ... }。即使你和我知道该query方法会调用RowCallbackHandler从序列返回的 before，Kotlin 编译器也无法知道这一点。绝不允许从序列本身以外的函数和对象生成序列值，因为无法知道它们将在何处或何时运行。

为了解决这个问题，我们需要引入一种不同类型的协程：它可以在等待调用时自行挂起RowCallbackHandler。

不幸的是，因为我们在这里讨论的是 JDBC，所以通过引入成熟的协程可能不会获得太多好处。在幕后，对数据库的调用将始终以阻塞方式进行，从而消除了很多好处。不尝试“流式传输”结果，而只是以一种无聊、老式的方式迭代它们，可能会更简单。但我们还是要探索一下各种可能性。

序列的问题

序列是为按需计算而设计的，并且不是异步的。他们不能等待其他异步操作，例如回调。序列构建器的yield函数只是在等待调用者检索下一个项目时挂起，并且它是序列允许调用的唯一挂起函数。delay如果您尝试在序列中使用简单的挂起调用，则可以演示这一点。您将收到一个编译错误，让您知道您正在受限的协程范围内运行。

sequence<String> { delay(1000) } // doesn't compile

如果无法调用挂起函数，就无法等待调用回调。认识到这一限制，Kotlin 为按需值流提供了一种替代机制，以异步方式提供数据。它被称为Flow。

回调流程

Roman Elizarov 在他的 Medium 文章Callbacks and Kotlin Flows中很好地描述了使用 Flows 从回调接口提供值的机制。

如果您确实想使用回调流程，只需将 替换sequence为callbackFlow，然后替换yield为sendBlocking。

您的代码可能如下所示：

fun findManyObjects(): Flow<Thing> = callbackFlow {
    val rowHandler = object : RowCallbackHandler {
        override fun processRow(resultSet: ResultSet) {
            val thing = // create from resultSet
            sendBlocking(thing)
        }
    }
    jdbcTemplate.query("select * from ...", rowHandler)
    close() // the query is finished, so there are no more rows
}

更简单的流程

虽然这是流式传输回调提供的值的惯用方法，但它可能不是解决此问题的最简单方法。通过完全避免回调，您可以使用更常见的flow构建器，将每个值传递给其emit函数。但现在你有了协程形式的异步，你不能只返回一个流，然后让 Spring 立即关闭结果集。您需要能够延迟结果集的关闭，直到流实际被消耗为止。这意味着剥离RowCallbackHandleror提供的抽象ResultSetExtractor，它期望以阻塞方式处理所有结果，而是提供您自己的实现。

fun Connection.findManyObjects(): Flow<Thing> = flow {
    prepareStatement("select * from ...").use { statement ->
        statement.executeQuery().use { resultSet ->        
            while (resultSet.next()) {
                val thing = // create from resultSet
                emit(thing)
            }
        }
    }
}

请注意这些use块，它们将处理关闭语句和结果集。因为在循环完成并且所有值都已发出use之前我们不会到达块的末尾，所以在结果集保持打开状态时流程可以自由挂起。while

那么为什么要使用流呢？

您可能会注意到，如果这样做，您实际上可以flow将and替换emit为sequenceand yield。那么我们已经回到原点了吗？嗯，有点像。区别在于 aflow只能从协程中使用，而使用sequence，您可以迭代结果值而不需要暂停。在这种特殊情况下，这是一个很难进行的调用，因为JDBC 操作总是阻塞的。

• 如果使用序列，调用线程将在等待接收数据时阻塞。序列中的值始终由消耗序列的事物计算，因此如果序列调用阻塞函数，使用者的线程将阻塞等待该值。在非协程应用程序中，这可能没问题，但如果您使用协程，您确实希望避免将阻塞调用隐藏在看似无害的序列中。
• 如果您使用流，则至少可以通过让流在特定调度程序上运行来隔离阻塞调用。例如，您可以使用内置 IO 调度程序来执行 JDBC 调用，然后切换回默认调度程序以进行进一步的处理。如果您确实想要流式传输值，我认为这是比使用序列更好的方法。

考虑到所有这些，如果您选择这些解决方案之一，则需要小心使用协程和调度程序。如果您不想担心这一点，那么使用常规ResultSetExtractor并暂时忘记序列和流程并没有什么问题。