光滑的AsyncExecutor异常

Posted 2021-05-04

我有一个用Scala 2.11编写的Play 2.5 + Slick 3.2应用程序，它使用了play-slick 2.1.1插件。

Play的线程池的配置是默认值。我知道这是Akka调度员。配置Slick的线程池：

slick.dbs.data.db.numThreads=75
slick.dbs.data.db.queueSize=1000

HikariCP连接池的配置也是默认配置。

有一个Play异步操作，它按组ID检索主题（数据库中约8000个主题），并且每个检索到的主题从其他两个引用的表中组装其DTO：

def getSubjects(groupId: Int) : Future[Seq[SubjectDTO]] = Action.async { request =>
  db.run(retrieveSubjects(request.groupId).withPinnedSession).flatMap { subjects => 
    Future.traverse(subjects)(subjectDTOBuilder())
  }
}

private def subjectDTOBuilder() =
  (subject: Subject) => {
    for {
      subjData <- db.run(getSubjectsAdditionalData(subject.id).withPinnedSession)
      subjInfo <- db.run(getSubjectsMoreInfo(subject.id).withPinnedSession)
    } yield SubjectDTO(subject, subjData, subjInfo)
  }

执行此Action时，它会在某些时候失败：

Task slick.basic.BasicBackend$DatabaseDef$$anon$2@286a41fb rejected from slick.util.AsyncExecutor$$anon$2$$anon$1@46d9be94[Running, pool size = 75, active threads = 75, queued tasks = 1000, completed tasks = 55419]"

当我重写subjectDTOBuilder时，异常消失了：

private def subjectDTOBuilder() =
  (subject: Subject) => {
    val future = db.run(getSubjectsAdditionalData(subject.id).withPinnedSession)
    val future2 = db.run(getSubjectsMoreInfo(subject.id).withPinnedSession)
    for {
      subjData <- future
      subjInfo <- future2
    } yield SubjectDTO(subject, subjData, subjInfo)
  }

我知道现在这些“无关”的期货并行执行，但我没有得到Slick异常的实际原因。 Slick的线程池有什么问题？有人可以解释这种异常可能发生的原因吗？

答案

错误消息显示“排队的任务= 1000”。默认任务队列长度也是1000.这意味着您可能一次调度了太多任务，因此任务执行程序的输入队列溢出。此队列包含已调度但仍在等待执行的所有任务。

因此，解决方案可能是同时安排较少数量的任务或增加队列长度。您可以通过提供自己的隐式ExecutionContext实例来配置此行为。