scala - 如何异步流式传输/分块数据库结果？

Question

我想使用 Enumerator(类似于 ScalaStream 示例)将数据库结果分 block 以作为 TSV 响应。我可能有数千行，我不想对结果进行分页，也不想将整个 ResultSet 累积到单个 String 中。如果需要，可以在行分隔符处不中断 block 。换句话说，唯一的标准是将结果分 block ，而不是它遵守 TSV 分隔符。

我想在我的 Action 中做这样的事情:

Ok.stream(new Iterator[ResultSet] {
  val conn = DB.getConnection()
  val stmt = conn.createStatement
  val rs = stmt.executeQuery("""select * from atable""")
  def hasNext = !rs.isLast() && !rs.isClosed()
  def next() = {
    if (!rs.next()) {
      conn.close()
      null
    } else rs
  }
}.toStream)

不幸的是，Ok.stream 需要一个 java.io.InputStream 并且将结果包装在 InputStream 中似乎有些过分. Ok.stream 也接受 Enumerator 但我不确定如何在这种情况下创建一个。

Answer 1

首先，您不能“异步”执行此操作，因为 JDBC API 是同步的 - 当您调用 rs.next() 时，它会同步阻塞。但这没关系，只需要确保您的线程池已调整为允许阻塞操作(即，使它们变大)。 Play 有一个异步流 API(它与您最可能习惯的同步 InputStream/OutputStream 完全不同)，它使用称为迭代器/枚举器的东西。您想创建一个枚举器来枚举您的 ResultSet。本质上，你想做一些看起来像这样的事情:

import play.api.libs.iteratee._

val conn = DB.getConnection()
val stmt = conn.createStatement
val resultSet = stmt.executeQuery("""select * from atable""")

Ok.stream(Enumerator.unfold(resultSet) { (rs: ResultSet) =>
  if (rs.next()) {
    val chunk = // Read the result from the ResultSet and format it in the way you want it formatted
    Some((rs, chunk))
  } else None
}.onDoneEnumerating {
  resultSet.close()
  stmt.close()
  conn.close()
})