3

我正在使用 anorm 访问我的数据库上的数据。数据库是使用另一个服务写入的，该服务是用 Java 制作的，并使用 ebean 持久化。

我有以下 scala 对象

import java.sql.Connection

import scala.concurrent.{ Future, blocking, future }
import scala.concurrent.ExecutionContext.Implicits.global

import anorm.{ SQL, SqlQuery, SqlRow, sqlToSimple, toParameterValue }
import play.api.Logger
import play.api.Play.current
import play.api.db.DB

object Queries {

  private val readDataSource: String = play.Configuration.root().getString("data.provider.api.source", "default")
  //better IO execution context

  import play.api.libs.concurrent.Execution.Implicits.defaultContext

  private val dataSetDescription: SqlQuery = SQL("SELECT DISTINCT platform, name FROM data_nugget")

  private val identityCreationTime: SqlQuery = SQL("SELECT i.creation_time FROM identity i WHERE platform = {pfm} AND userid = {uid};")

  private val identityData: SqlQuery = SQL("SELECT n.name, n.value FROM data_nugget n WHERE platform = {pfm} AND userid = {uid};")

  private val playerData: SqlQuery = SQL("SELECT n.platform, n.name, n.value, r.userid, r.registration_time FROM data_nugget n JOIN registration r ON n.platform=r.platform AND n.userid=r.userid  WHERE r.playerid = {pid} AND r.application = {app};")

  private def withAsyncAnormConnection(function: Connection => Stream[SqlRow]): Future[List[SqlRow]] = {
    future {
      blocking {
        DB.withConnection(readDataSource)(c => function(c)).toList
      }
    }
  }

  def fetchDistinctDataNames(): Future[List[SqlRow]] = {
    withAsyncAnormConnection(implicit c => dataSetDescription())
  }

  def fetchIdentityCreationTime(platform: String, userid: String): Future[List[SqlRow]] = {
    withAsyncAnormConnection(implicit c => identityCreationTime.on("pfm" -> platform, "uid" -> userid)())
  }

  def fetchIdentityData(platform: String, userid: String): Future[List[SqlRow]] = {
    withAsyncAnormConnection(implicit c => identityData.on("pfm" -> platform, "uid" -> userid)())
  }

  def fetchRegistrationData(game: String, playerid: String): Future[List[SqlRow]] = {
    withAsyncAnormConnection(implicit c => playerData.on("app" -> game, "pid" -> playerid)())
  }

}

我使用它将我的 SQL 查询执行包装在期货中。

每次我运行任何这些查询时，都会收到以下堆栈跟踪错误：

(Error,com.mysql.jdbc.SQLError.createSQLException(SQLError.java:1073)
com.mysql.jdbc.SQLError.createSQLException(SQLError.java:987)
com.mysql.jdbc.SQLError.createSQLException(SQLError.java:982)
com.mysql.jdbc.SQLError.createSQLException(SQLError.java:927)
com.mysql.jdbc.ResultSetImpl.checkClosed(ResultSetImpl.java:794)
com.mysql.jdbc.ResultSetImpl.next(ResultSetImpl.java:7139)
anorm.Sql$$anonfun$resultSetToStream$1.apply(Anorm.scala:527)
anorm.Sql$$anonfun$resultSetToStream$1.apply(Anorm.scala:527)
anorm.Useful$.unfold(Anorm.scala:315)
anorm.Useful$$anonfun$unfold$1.apply(Anorm.scala:317)
anorm.Useful$$anonfun$unfold$1.apply(Anorm.scala:317)
scala.collection.immutable.Stream$Cons.tail(Stream.scala:1078)
scala.collection.immutable.Stream$Cons.tail(Stream.scala:1070)
scala.collection.immutable.Stream.foreach(Stream.scala:548)
scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
scala.collection.mutable.ListBuffer.$plus$plus$eq(ListBuffer.scala:178)
scala.collection.mutable.ListBuffer.$plus$plus$eq(ListBuffer.scala:45)
scala.collection.TraversableLike$class.to(TraversableLike.scala:629)
scala.collection.AbstractTraversable.to(Traversable.scala:105)
scala.collection.TraversableOnce$class.toList(TraversableOnce.scala:243)
scala.collection.AbstractTraversable.toList(Traversable.scala:105)
controllers.dataprovider.data.Queries$$anonfun$withAsyncAnormConnection$1$$anonfun$apply$1.apply(Queries.scala:31)
controllers.dataprovider.data.Queries$$anonfun$withAsyncAnormConnection$1$$anonfun$apply$1.apply(Queries.scala:31)
scala.concurrent.impl.ExecutionContextImpl$DefaultThreadFactory$$anon$2$$anon$3.block(ExecutionContextImpl.scala:44)
scala.concurrent.forkjoin.ForkJoinPool.managedBlock(ForkJoinPool.java:2803)
scala.concurrent.impl.ExecutionContextImpl$DefaultThreadFactory$$anon$2.blockOn(ExecutionContextImpl.scala:41)
scala.concurrent.package$.blocking(package.scala:50)
controllers.dataprovider.data.Queries$$anonfun$withAsyncAnormConnection$1.apply(Queries.scala:30)
controllers.dataprovider.data.Queries$$anonfun$withAsyncAnormConnection$1.apply(Queries.scala:30)
scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)
scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
scala.concurrent.forkjoin.ForkJoinTask$AdaptedRunnableAction.exec(ForkJoinTask.java:1417)
scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:262)
scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:975)
scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1478)
scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:104))

我已经在使用 jdbc 的以前的 Java 服务上遇到过这些，但在这里我没有触及 ResultSet，我什至从我从连接接收到的行流中尽快返回一个列表。

怎么了？我在哪里关闭 ResultSet？我重构错了什么？

作为说明，在此服务的原型上（当一切都在控制器中时），我曾经在代码中直接使用 SQL("...") ，如下所示：

future {
    blocking {
      DB.withConnection(implicit c => {
        SQL("SELECT DISTINCT platform, name FROM data_nugget")().map(row => (row[String]("platform"), row[String]("name"))).toArray
      })
    }
  }

它工作得很好。

PS：很抱歉堆栈跟踪和代码的长复制/粘贴......试图详细说明。

4

2 回答 2

3

我自己解决了它，这是一条非常好的路线。

我改变了这个功能

  private def withAsyncAnormConnection(function: Connection => Stream[SqlRow]): Future[List[SqlRow]] = {
    future {
      blocking {
        DB.withConnection(readDataSource)(c => function(c)).toList
      }
    }
  }

对此：

  private def withAsyncAnormConnection(function: Connection => Stream[SqlRow]): Future[List[SqlRow]] = {
    future {
      blocking {
        DB.withConnection(readDataSource)(c => function(c).toList)
      }
    }
  }

诀窍是我使用的是“贷款模式” withConnection，所以我需要在释放连接之前遍历Stream以获取所有行。
连接仅在此圆括号内有效(c => function(c).toList)

于 2013-05-14T10:06:17.553 回答
2

为您工作的代码和不工作的代码之间是有区别的。在您的工作示例中，您正在调用实例map的惰性Stream。Row在非工作示例中，您在toList不使用map. 也许是在强制对块内map的底层证券进行全面处理，而不是，让它变得懒惰，直到你离开块，然后底层证券被关闭。也许你可以修改你的新代码来尝试映射结果（映射到自身，没有实际的映射逻辑），看看这是否解决了任何问题。ResultSetwithConnectiontoListwithConnectionResultSetRow

于 2013-05-14T01:08:47.420 回答