批量插入时 Slick 3.0 中的数据库异常

Posted 2023-04-18

【中文标题】批量插入时 Slick 3.0 中的数据库异常

【英文标题】：Database Exception in Slick 3.0 while batch insert

【发布时间】：2015-06-11 06:06:36

【问题描述】：

在 slick 3 中通过批量插入每五秒插入数千条记录时，我得到了

org.postgresql.util.PSQLException: FATAL: sorry, too many clients already

我的数据访问层如下所示：

val db: CustomPostgresDriver.backend.DatabaseDef = Database.forURL(url, user=user, password=password, driver= jdbcDriver)



 override def insertBatch(rowList: List[T#TableElementType]): Future[Long] = 
    val res = db.run(insertBatchQuery(rowList)).map(_.head.toLong).recover case ex:Throwable=> RelationalRepositoryUtility.handleBatchOperationErrors(ex)
//db.close()
        res
      

  override def insertBatchQuery(rowList: List[T#TableElementType]): FixedSqlAction[Option[Int], NoStream, Write] = 
    query ++= (rowList)
  

在插入批处理中关闭连接没有效果...它仍然给出相同的错误。

我正在从我的代码中调用插入批处理，如下所示：

val temp1 = list1.flatMap  li =>
        Future.sequence(li.map  trip =>
            val data = for 
              tripData <- TripDataRepository.insertQuery( trip.tripData)
              subTripData <- SubTripDataRepository.insertBatchQuery(getUpdatedSubTripDataList(trip.subTripData, tripData.id))
             yield ((tripData, subTripData))
            val res=db.run(data.transactionally)
          res
//db.close()
        )
      

如果我在这里工作后关闭连接，正如您在注释代码中看到的那样，我会收到错误：

java.util.concurrent.RejectedExecutionException: Task slick.backend.DatabaseComponent$DatabaseDef$$anon$2@6c3ae2b6 rejected from java.util.concurrent.ThreadPoolExecutor@79d2d4eb[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 1]

像这样调用没有 Future.sequence 的方法后：

 val temp1 =list.map  trip =>
          val data = for 
            tripData <- TripDataRepository.insertQuery( trip.tripData)
            subTripData <- SubTripDataRepository.insertBatchQuery(getUpdatedSubTripDataList(trip.subTripData, tripData.id))
           yield ((tripData, subTripData))
          val res=db.run(data.transactionally)
          res
      

我仍然收到太多客户错误...

【问题讨论】：

“抱歉，已经有太多客户端”表示您打开了很多很多连接，但您从未关闭它们。

你能发布你是如何调用 insertBatch 和更多令人痛心的代码吗？正如@a_horse_with_no_name 所建议的那样，该错误意味着您打开了太多的连接。

li.map trip => val data = for tripData obj) + = trip.tripData) subTripData

这就是我调用插入批处理方法的方式......即使我在完成工作后尝试在插入批处理中关闭连接......它仍然是同样的错误

你能验证你的 postgres 数据库允许多少个连接吗？请在 postgres "show max_connections" 上使用此命令

【参考方案1】：

这个问题的根源在于您同时创建了一个无限的Future 列表，每个都连接到数据库 - list 中的每个条目一个。

这可以通过串行运行插入来解决，强制每个插入批次依赖于前一个：

// Empty Future for the results. Replace Unit with the correct type - whatever
// "res" is below.
val emptyFuture = Future.successful(Seq.empty[Unit])
// This will only insert one at a time. You could use list.sliding to batch the
// inserts if that was important.
val temp1 = list.foldLeft(emptyFuture)  (previousFuture, trip) =>
  previousFuture flatMap  previous =>
    // Inner code copied from your example.
    val data = for 
      tripData <- TripDataRepository.insertQuery(trip.tripData)
      subTripData <- SubTripDataRepository.insertBatchQuery(getUpdatedSubTripDataList(trip.subTripData, tripData.id))
     yield ((tripData, subTripData))
    val res = db.run(data.transactionally)
    previous :+ res