database in Slick 3.0 mentre lotti inserire

Question

Mentre l'inserimento di migliaia di record per cinque secondi attraverso inserto batch nella chiazza di petrolio 3 sto ottenendo

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

Il mio livello di accesso ai dati si presenta come:

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) 
    } 

chiusura la connessione in insert batch non ha alcun effetto ... dà ancora lo stesso errore.

sto chiamando inserto in batch dal mio codice come questo:

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() 
     }) 
     } 

se chiudo la connessione dopo il mio lavoro qui, come si può vedere nella commentato codice ottengo errore:

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] 

Dopo chiamando il metodo senza Future.sequence in questo modo:

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 
     } 

Ho ancora troppi client errore ...

Answer 1

La radice di questo problema è che si sta eseguendo un elenco illimitato di Future contemporaneamente, ciascuna connessione al database, una per voce in list.

Questo può essere risolto eseguendo i vostri inserti in serie, costringendo ogni lotto inserto a dipendere dalla precedente:

// 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 
    } 
}