在“大型” Future.sequence

如何解决在“大型” Future.sequence

我一整天都在想办法解决这个问题。

目的是将多个字符串序列插入到表的单个列中。

我有一个这样的方法：

case class Column(strings: Seq[String])

def insertColumns(columns: Seq[Column]) = for {
_ <- Future.sequence(columns.map(col => insert(col)))
} yield()

private def insert(column: Column) =
  db.run((stringTable ++= rows)) //slick batch insert

这在一定程度上起作用了。 我测试了 2100 列的序列（每列有 100 个字符串），它工作正常。 但是一旦我将列数增加到 3100+，就会出现这个错误

Task slick.basic.BasicBackend$DatabaseDef$$anon$3@293ce053 rejected from slick.util.AsyncExecutor$$anon$1$$anon$2@3e423930[Running,pool size = 10,active threads = 10,queued tasks = 1000,completed tasks = 8160]

我在几个地方读到过这样做会有所帮助

case class Column(strings: Seq[String])

val f = Future.sequence(columns.map(col => insert(col)))

def insertColumns(columns: Seq[Column]) = for {
_ <- f
} yield()

private def insert(column: Column) =
  db.run((stringTable ++= rows)) //slick batch insert

没有。

我在 insert 中尝试了几种更改组合

Future.sequence(
rows.grouped(500).toSeq.map(group => db.run(DBIO.seq(stringTable ++= group)))
)

Source(rows).buffer(500,OverflowStrategy.backpressure)
  .via(
    Slick.flow(row => stringTable += row)
  )
  .log("nr-of-inserted-rows")
  .runWith(Sink.ignore)

Source(rows)
.runWith(Slick.sink(1,row => stringTable += row))

我试过了：

• 不要在我的配置中使用 reWriteBatchedInserts=true
• (dataColumnStringsTable ++= rows).transactionally 选项
• 使用特定的执行上下文启用单个线程：implicit val ec: ExecutionContext = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(1)) 尝试按顺序执行期货

除了重新处理我的订阅者以接收和阻止我的消息（字符串序列）并处理队列消息传递端的背压之外，我没有任何其他想法。

我正在使用 slick（带有 alpakka-slick）3.3.3 / HikariCP 3.2.0 / Postgres 13.2

我的配置是这样

slick {
  profile = "slick.jdbc.PostgresProfile$"
  db {
      connectionPool = "HikariCP"
      dataSourceClass = "slick.jdbc.DriverDataSource"
      properties = {
        driver = "org.postgresql.Driver"
        user = "postgres"
        password = "password"
        url = "jdbc:postgresql://"${slick.db.host}":5432/slick?reWriteBatchedInserts=true"
      }
      host = "localhost"
      numThreads = 10
      maxConnections = 100
      minConnections = 1
    }
}

感谢您的帮助。

解决方法

您不应将 Future.sequence 用于包含多个元素的集合。每个 Future 都是在后台运行的计算。因此，当您为 3000 columns 的集合运行此程序时：

Future.sequence(columns.map(col => insert(col)))

您一次有效地产生了 3000 个操作。结果，执行器可能会开始拒绝新任务。

解决方案是使用 Akka Streams 处理输入集合。在您的情况下，这意味着从 Source（而不是从 columns）创建 rows。这将确保执行器不会被太多的并行操作淹没。我没有使用过 alpakka-slick，但查看 docs，解决方案应该如下所示：

Source(columns)
  .via(
    Slick.flow(column => stringTable ++= column.rows) 
  )
  // further processing here

此外，如果“列”来自消息队列，您甚至可能不需要中间的 Seq[Column]。您可能只需要定义从队列中读取的 Source 的 Column，并使用 Slick 流对其进行处理。