Scala 中带有 Play 框架的迭代器的分块响应

Question

我有一个来自数据库调用的大型结果集，我需要将其流回给用户，因为它不能全部放入内存。

我可以通过设置选项从数据库中返回结果

val statement = session.conn.prepareStatement(query, 
                java.sql.ResultSet.TYPE_FORWARD_ONLY,
                java.sql.ResultSet.CONCUR_READ_ONLY)
statement.setFetchSize(Integer.MIN_VALUE)
....
....
val res = statement.executeQuery

然后使用迭代器

val result = new Iterator[MyResultClass] {
    def hasNext = res.next
    def next = MyResultClass(someValue = res.getString("someColumn"), anotherValue = res.getInt("anotherValue"))
}

在 Scala 中，Iterator 扩展了 TraversableOnce，这应该允许我根据 https://www.playframework.com/documentation/2.3.x/ScalaStream 的文档将 Iterator 传递给用于播放框架中的 Chunked Response 的 Enumerator 类

查看 Enumerator 的源代码时，我发现它有一个重载的 apply 方法来使用 TraversableOnce 对象

我尝试使用以下代码

import play.api.libs.iteratee.Enumerator
val dataContent = Enumerator(result)
Ok.chunked(dataContent)

但这不起作用，因为它会引发以下异常

Cannot write an instance of Iterator[MyResultClass] to HTTP response. Try to define a Writeable[Iterator[MyResultClass]]

我在文档中找不到关于 Writable 是什么或做什么的任何地方。我认为一旦 Enumerator 使用了 TraversableOnce 对象，它就会从那里获取它，但我猜不是？？

Answer 1

您的方法存在问题

你的方法有两个问题：

1. 您正在将Iterator 写入Enumerator / Iteratee。你应该写Iterator的内容而不是整个Iterator
2. Scala 不知道如何在 HTTP 流上表达 MyResultClass 的对象。在编写它们之前尝试将它们转换为 String 表示（例如 JSON）。

示例

build.sbt

一个简单的 Play Scala 项目，支持 H2 和 SQL。

lazy val root = (project in file(".")).enablePlugins(PlayScala)

scalaVersion := "2.11.6"

libraryDependencies ++= Seq(
  jdbc,
  "org.scalikejdbc" %% "scalikejdbc"       % "2.2.4",
  "com.h2database"  %  "h2"                % "1.4.185",
  "ch.qos.logback"  %  "logback-classic"   % "1.1.2"
)

项目/plugins.sbt

只是当前稳定版本中 sbt play 插件的最小配置

resolvers += "Typesafe repository" at "http://repo.typesafe.com/typesafe/releases/"

addSbtPlugin("com.typesafe.play" % "sbt-plugin" % "2.3.8")

配置/路由

/json 上只有一条路线

GET    /json                        controllers.Application.json

Global.scala

配置文件，在 Play 应用程序启动期间创建并使用演示数据填充数据库

import play.api.Application
import play.api.GlobalSettings
import scalikejdbc._

object Global extends GlobalSettings {

  override def onStart(app : Application): Unit = {

    // initialize JDBC driver & connection pool
    Class.forName("org.h2.Driver")
    ConnectionPool.singleton("jdbc:h2:mem:hello", "user", "pass")

    // ad-hoc session provider
    implicit val session = AutoSession


    // Create table
    sql"""
      CREATE TABLE persons (
        customer_id SERIAL NOT NULL PRIMARY KEY,
        first_name VARCHAR(64),
        sure_name VARCHAR(64)
      )""".execute.apply()

    // Fill table with demo data
    Seq(("Alice", "Anderson"), ("Bob", "Builder"), ("Chris", "Christoph")).
      foreach { case (firstName, sureName) =>
        sql"INSERT INTO persons (first_name, sure_name) VALUES (${firstName}, ${sureName})".update.apply()
    }
  }
}

模型/Person.scala

在这里，我们定义了数据库模式和数据库对象的 Scala 表示。这里的关键是函数personWrites。它将 Person 对象转换为 JSON 表示（真正的代码方便地由宏生成）。

package models

import scalikejdbc._
import scalikejdbc.WrappedResultSet
import play.api.libs.json._

case class Person(customerId : Long, firstName: Option[String], sureName : Option[String])

object PersonsTable extends SQLSyntaxSupport[Person] {
  override val tableName : String = "persons"
  def apply(rs : WrappedResultSet) : Person =
    Person(rs.long("customer_id"), rs.stringOpt("first_name"), rs.stringOpt("sure_name"))
}

package object models {
  implicit val personWrites: Writes[Person] = Json.writes[Person]
}

控制器/Application.scala

这里有 Iteratee / Enumerator 代码。首先我们从数据库中读取数据，然后我们将结果转换为迭代器，然后转换为枚举器。该 Enumerator 不会有用，因为它的内容是 Person 对象，而 Play 不知道如何通过 HTTP 编写此类对象。但是在personWrites 的帮助下，我们可以将这些对象转换为 JSON。 Play 知道如何通过 HTTP 编写 JSON。

package controllers

import play.api.libs.json.JsValue
import play.api.mvc._
import play.api.libs.iteratee._
import scala.concurrent.ExecutionContext.Implicits.global
import scalikejdbc._

import models._
import models.personWrites

object Application extends Controller {

  implicit val session = AutoSession

  val allPersons : Traversable[Person] = sql"SELECT * FROM persons".map(rs => PersonsTable(rs)).traversable().apply()
  def personIterator(): Iterator[Person] = allPersons.toIterator
  def personEnumerator() : Enumerator[Person] = Enumerator.enumerate(personIterator)
  def personJsonEnumerator() : Enumerator[JsValue] = personEnumerator.map(personWrites.writes(_))

  def json = Action {
    Ok.chunked(personJsonEnumerator())
  }
}

讨论

数据库配置

在这个例子中，数据库配置是一个 hack。通常我们会配置 Play，使其提供数据源并在后台处理所有数据库内容。

JSON 转换

在代码中我直接调用 JSON 转换。有更好的方法，导致代码更紧凑（但对初学者来说更容易理解）。

您得到的响应不是真正有效的 JSON。示例：

{"customerId":1,"firstName":"Alice","sureName":"Anderson"}
{"customerId":2,"firstName":"Bob","sureName":"Builder"}
{"customerId":3,"firstName":"Chris","sureName":"Christoph"}

（备注：换行符仅用于格式化。在电线上看起来像这样：

...son"}{"custom...

相反，您会将有效 JSON 块分块在一起。那是你要求的。接收端可以自己消费每个块。但是有一个问题：你必须想办法把响应分成有效的块。

请求本身确实是分块的。考虑以下 HTTP 标头（JSON HAR 格式，从 Google Chrome 导出）：

     "status": 200,
      "statusText": "OK",
      "httpVersion": "HTTP/1.1",
      "headers": [
        {
          "name": "Transfer-Encoding",
          "value": "chunked"
        },
        {
          "name": "Content-Type",
          "value": "application/json; charset=utf-8"
        }

代码组织

我在控制器中放了一些 SQL 代码。在这种情况下，这完全没问题。如果代码变大，模型中的 SQL 东西可能会更好，让控制器使用更通用的（在这种情况下：“monadic plus”，即map、filter、flatMap）接口。

在控制器中 JSON 代码和 SQL 代码混合在一起。当代码变大时，你应该组织它，例如每个技术或每个模型对象/业务领域。

阻塞迭代器

使用迭代器会导致阻塞行为。这通常是一个大问题，但对于必须有大量负载（每秒数百或数千次点击）或必须快速响应（想想在堆栈交换上实时工作的交易算法）的应用程序应该避免这种情况。在这种情况下，您可以使用 NoSQL 数据库作为缓存（请不要将其用作唯一的数据存储）或非阻塞 JDBC（例如async postgres / mysql）。再说一遍：这对于大型应用程序来说不是必需的。

注意：一旦您转换为迭代器，请记住您只能使用一次迭代器。对于每个请求，您都需要一个新的迭代器。

结论

一个完整的 WebApp，包括数据库访问完全在一个（不是那么短的）SO 答案中。我真的很喜欢 Play 框架。

此代码用于教育目的。在某些地方特别尴尬，以使初学者更容易理解这些概念。在实际的应用程序中，您会理顺这些事情，因为您已经了解这些概念并且您只想查看代码的用途（为什么会出现？它使用了哪些工具？它什么时候做什么？）乍一看。

玩得开心！